yml2213
df8c63b829
协作台账：新表 huya_external_orders 替代 Excel 多人协同表（设计见
docs/设计-虎牙外部CK协作台账.md），一条 CK 一行，所有人共享、改动留痕。
- 前端 pages/HuyaCollabPage.tsx + 菜单「虎牙 → 外部单协作」/huya/collab：
内联编辑（带乐观锁 version）、图片缩略图、详情抽屉（字段表单 + 两个图片槽位 +
操作时间线）、粘贴导入弹窗（带 TSV 预览）、批量/单条验证 CK、页头统计条、筛选
- 两套状态机：业务 pending→bound→recharged→scripted→done（旁路 void）、
凭据 unverified/valid/invalid/disabled；唯一写入口在 services/huya_collab_service.py，
字段级审计 + 整数 version 乐观锁（不用 updated_at：DATETIME 只到秒）
- CK 只读探针 check_user_bind_game_account（真机判据：响应非 None 且 status == 200），
成功即覆盖写绑定快照 bind_game_account/bind_game_role/bind_game_channel（迁移 0056），
失败只改凭据状态、不动业务状态；人工只能停用/恢复（POST /{id}/ck/status）
- 贴/换/重建 CK、TSV 批量导入（SAVEPOINT 单行隔离，部分失败不整体失败）、
权限 huya:collab:view/edit/manage（删除只给运营）
- 孤儿图清理脚本 scripts/cleanup_orphan_file_assets.py（默认 --dry-run）
图片上传（协作台账截图用）：OSS/本地双后端 + 缩略图变体 + /api/files 代理回读，
权限 file:upload / file:view。
验证：pytest 880 passed / 6 skipped；ruff format 全绿、ruff check 仅剩既有告警；
前端 tsc -b / eslint / vite build 通过；接口端到端冒烟 29 项全绿（TestClient，探针用假客户端）；
探针真机冒烟 3 例（有效 CK=True、别人 uid=False、裸 cookie=False）。
410 lines
16 KiB
Python
410 lines
16 KiB
Python
"""在线协作图片上传：存储层 + 服务层的行为契约。
|
||
|
||
只跑本地目录后端（``OSS_MODE=local`` + 临时目录），不依赖 OSS。
|
||
覆盖：魔数嗅探、scene 归一化、object key 布局、变体生成、落库、
|
||
超限/类型拒绝、失败补偿删除、变体缺失回退原图、key 路径穿越。
|
||
"""
|
||
|
||
from __future__ import annotations
|
||
|
||
import io
|
||
import re
|
||
from pathlib import Path
|
||
from typing import Any
|
||
from urllib.parse import parse_qs, urlsplit
|
||
|
||
import pytest
|
||
from PIL import Image
|
||
|
||
from tests.support.db import make_test_session
|
||
from web.backend.models import FileAsset
|
||
from web.backend.services.file_service import (
|
||
FileUploadError,
|
||
build_object_key,
|
||
generate_image_variants,
|
||
get_stored_object,
|
||
image_variant_fallback_keys,
|
||
image_variant_key,
|
||
normalize_scene,
|
||
resolve_content_type,
|
||
sniff_content_type,
|
||
upload_file_asset,
|
||
)
|
||
from web.backend.services.file_storage import (
|
||
S3ObjectStorage,
|
||
StorageUnavailableError,
|
||
get_object_storage,
|
||
load_storage_settings,
|
||
reset_object_storage_cache,
|
||
)
|
||
|
||
|
||
@pytest.fixture
|
||
def local_storage(tmp_path: Path, monkeypatch: pytest.MonkeyPatch) -> Path:
|
||
"""把对象存储指到临时目录，并在用例结束后清掉单例缓存。"""
|
||
monkeypatch.setenv("OSS_MODE", "local")
|
||
monkeypatch.setenv("OSS_LOCAL_ROOT", str(tmp_path))
|
||
monkeypatch.setenv("OSS_MAX_UPLOAD_SIZE_MB", "10")
|
||
reset_object_storage_cache()
|
||
yield tmp_path
|
||
reset_object_storage_cache()
|
||
|
||
|
||
def _png_bytes(width: int = 1200, height: int = 800) -> bytes:
|
||
buffer = io.BytesIO()
|
||
Image.new("RGB", (width, height), (30, 90, 200)).save(buffer, "PNG")
|
||
return buffer.getvalue()
|
||
|
||
|
||
def _jpeg_bytes(width: int = 1600, height: int = 900) -> bytes:
|
||
buffer = io.BytesIO()
|
||
Image.new("RGB", (width, height), (200, 90, 30)).save(buffer, "JPEG")
|
||
return buffer.getvalue()
|
||
|
||
|
||
class _FailingSession:
|
||
"""commit 必炸的假会话，用来验证对象补偿删除。"""
|
||
|
||
def __init__(self) -> None:
|
||
self.added: list[Any] = []
|
||
self.rolled_back = False
|
||
|
||
def add(self, obj: Any) -> None:
|
||
self.added.append(obj)
|
||
|
||
def commit(self) -> None:
|
||
raise RuntimeError("database is down")
|
||
|
||
def rollback(self) -> None:
|
||
self.rolled_back = True
|
||
|
||
|
||
class TestContentTypeAndScene:
|
||
def test_sniff_content_type_by_magic_bytes(self):
|
||
assert sniff_content_type(_png_bytes(4, 4)) == "image/png"
|
||
assert sniff_content_type(_jpeg_bytes(4, 4)) == "image/jpeg"
|
||
assert sniff_content_type(b"%PDF-1.7\n") == "application/pdf"
|
||
assert sniff_content_type(b"hello world") == ""
|
||
|
||
def test_magic_bytes_win_over_lying_content_type(self):
|
||
"""前端声称是 png、实际是 jpeg 时，以文件头为准。"""
|
||
assert resolve_content_type("image/png", _jpeg_bytes(4, 4)) == "image/jpeg"
|
||
# 识别不出来时才回退到调用方给的 Content-Type。
|
||
assert resolve_content_type("application/pdf", b"not-an-image") == (
|
||
"application/pdf"
|
||
)
|
||
|
||
def test_normalize_scene_keeps_only_safe_chars(self):
|
||
assert normalize_scene("Collab Image") == "collab-image"
|
||
assert normalize_scene("../../etc/passwd") == "etc-passwd"
|
||
assert normalize_scene("") == "misc"
|
||
assert normalize_scene("协作图片") == "misc"
|
||
|
||
def test_object_key_layout_is_unpredictable_and_date_partitioned(self):
|
||
key = build_object_key("collab", "image/png", "screenshot.png")
|
||
assert re.fullmatch(r"collab/\d{4}/\d{2}/\d{2}/[0-9a-f]{24}\.png", key)
|
||
assert build_object_key("collab", "image/png", "a.png") != key
|
||
|
||
|
||
class TestImageVariants:
|
||
def test_png_source_keeps_alpha_and_outputs_webp(self):
|
||
variants = generate_image_variants(_png_bytes(), "image/png")
|
||
names = {variant.name: variant for variant in variants}
|
||
assert set(names) == {"thumb", "medium"}
|
||
for variant in names.values():
|
||
assert variant.content_type == "image/webp"
|
||
with Image.open(io.BytesIO(variant.content)) as image:
|
||
assert max(image.size) <= (480 if variant.name == "thumb" else 1280)
|
||
|
||
def test_jpeg_source_outputs_jpeg_and_shrinks_to_max_side(self):
|
||
variants = generate_image_variants(_jpeg_bytes(1600, 900), "image/jpeg")
|
||
thumb = next(item for item in variants if item.name == "thumb")
|
||
assert thumb.content_type == "image/jpeg"
|
||
with Image.open(io.BytesIO(thumb.content)) as image:
|
||
assert image.size == (480, 270)
|
||
|
||
def test_pdf_has_no_variants_and_broken_image_is_skipped(self):
|
||
assert generate_image_variants(b"%PDF-1.7\n", "application/pdf") == []
|
||
assert generate_image_variants(b"\x89PNG\r\n\x1a\nbroken", "image/png") == []
|
||
|
||
def test_variant_fallback_candidates_end_with_original(self):
|
||
candidates = image_variant_fallback_keys("collab/x.png", "thumb")
|
||
assert candidates == [
|
||
"collab/x.thumb.webp",
|
||
"collab/x.thumb.jpg",
|
||
"collab/x.thumb.png",
|
||
"collab/x.png",
|
||
]
|
||
# 无关的 variant 名不做候选展开。
|
||
assert image_variant_fallback_keys("collab/x.png", "huge") == ["collab/x.png"]
|
||
assert image_variant_key("collab/x.png", "medium", ".jpg") == (
|
||
"collab/x.medium.jpg"
|
||
)
|
||
|
||
|
||
class TestUploadFileAsset:
|
||
def test_upload_writes_objects_and_asset_row(self, local_storage: Path):
|
||
session = make_test_session()
|
||
try:
|
||
result = upload_file_asset(
|
||
session,
|
||
data=_png_bytes(1200, 800),
|
||
filename="截图.png",
|
||
content_type="image/png",
|
||
scene="collab",
|
||
uploader_type="user",
|
||
uploader_id="7",
|
||
)
|
||
finally:
|
||
session.close()
|
||
|
||
assert result.content_type == "image/png"
|
||
assert result.object_key.startswith("collab/")
|
||
# key 里的斜杠按两个参考实现的做法转义成 %2F，回读时能原样还原。
|
||
assert "collab%2F" in result.url
|
||
assert parse_qs(urlsplit(result.url).query)["key"][0] == result.object_key
|
||
# 与两个参考实现一致：变体 URL 直接指向变体对象，不再带 variant 参数。
|
||
thumb_key = parse_qs(urlsplit(result.thumbnail_url).query)["key"][0]
|
||
medium_key = parse_qs(urlsplit(result.medium_url).query)["key"][0]
|
||
assert thumb_key == f"{result.object_key[:-4]}.thumb.webp"
|
||
assert medium_key == f"{result.object_key[:-4]}.medium.webp"
|
||
|
||
# 原图 + 两个变体都真的落到对象存储里了。
|
||
stored = sorted(
|
||
path.name for path in local_storage.rglob("*") if path.is_file()
|
||
)
|
||
assert len(stored) == 3
|
||
assert any(name.endswith(".thumb.webp") for name in stored)
|
||
assert any(name.endswith(".medium.webp") for name in stored)
|
||
|
||
session = make_test_session()
|
||
try:
|
||
row = session.query(FileAsset).one()
|
||
assert row.object_key == result.object_key
|
||
assert row.scene == "collab"
|
||
assert row.uploader_id == "7"
|
||
assert row.size_bytes == len(_png_bytes(1200, 800))
|
||
assert row.thumbnail_object_key.endswith(".thumb.webp")
|
||
assert row.medium_object_key.endswith(".medium.webp")
|
||
assert row.status == "active"
|
||
finally:
|
||
session.close()
|
||
|
||
def test_pdf_upload_has_no_variants_and_falls_back_to_original(
|
||
self, local_storage: Path
|
||
):
|
||
session = make_test_session()
|
||
try:
|
||
result = upload_file_asset(
|
||
session,
|
||
data=b"%PDF-1.7\nfake",
|
||
filename="材料.pdf",
|
||
content_type="application/pdf",
|
||
scene="collab",
|
||
)
|
||
finally:
|
||
session.close()
|
||
|
||
assert result.thumbnail_url == result.url
|
||
stored = [path.name for path in local_storage.rglob("*") if path.is_file()]
|
||
assert len(stored) == 1
|
||
|
||
# 请求 thumb 但变体不存在时，回读回退到原图。
|
||
obj = get_stored_object(result.object_key, "thumb")
|
||
try:
|
||
assert obj.reader.read() == b"%PDF-1.7\nfake"
|
||
finally:
|
||
obj.reader.close()
|
||
|
||
def test_upload_rejects_oversize_file(self, tmp_path: Path, monkeypatch):
|
||
monkeypatch.setenv("OSS_MODE", "local")
|
||
monkeypatch.setenv("OSS_LOCAL_ROOT", str(tmp_path))
|
||
monkeypatch.setenv("OSS_MAX_UPLOAD_SIZE_MB", "1")
|
||
reset_object_storage_cache()
|
||
session = make_test_session()
|
||
try:
|
||
with pytest.raises(FileUploadError) as excinfo:
|
||
upload_file_asset(
|
||
session, data=b"x" * (1024 * 1024 + 1), filename="big.png"
|
||
)
|
||
finally:
|
||
session.close()
|
||
reset_object_storage_cache()
|
||
|
||
assert excinfo.value.status_code == 400
|
||
assert excinfo.value.code == "file_too_large"
|
||
|
||
def test_upload_rejects_unsupported_type(self, local_storage: Path):
|
||
session = make_test_session()
|
||
try:
|
||
with pytest.raises(FileUploadError) as excinfo:
|
||
upload_file_asset(session, data=b"#!/bin/sh\necho hi", filename="x.sh")
|
||
finally:
|
||
session.close()
|
||
assert excinfo.value.code == "file_type_invalid"
|
||
assert list(local_storage.rglob("*")) == []
|
||
|
||
def test_upload_rejects_empty_file(self, local_storage: Path):
|
||
session = make_test_session()
|
||
try:
|
||
with pytest.raises(FileUploadError) as excinfo:
|
||
upload_file_asset(session, data=b"", filename="empty.png")
|
||
finally:
|
||
session.close()
|
||
assert excinfo.value.code == "file_required"
|
||
|
||
def test_upload_removes_objects_when_db_write_fails(self, local_storage: Path):
|
||
"""写库失败必须把已写入的对象删掉，不能留下孤儿对象。"""
|
||
failing = _FailingSession()
|
||
with pytest.raises(RuntimeError, match="database is down"):
|
||
upload_file_asset(
|
||
failing,
|
||
data=_png_bytes(600, 400),
|
||
filename="shot.png",
|
||
content_type="image/png",
|
||
scene="collab",
|
||
)
|
||
|
||
assert failing.rolled_back is True
|
||
leftovers = [path for path in local_storage.rglob("*") if path.is_file()]
|
||
assert leftovers == [], f"补偿删除没清干净: {leftovers}"
|
||
|
||
|
||
class TestStorageSettings:
|
||
def test_file_asset_out_serializes_every_field(self):
|
||
"""坑 #2：Pydantic 会静默丢字段 —— 断言序列化结果而不是属性。"""
|
||
from web.backend.schemas import FileAssetOut
|
||
|
||
payload = FileAssetOut(
|
||
object_key="collab/2026/10/16/abc.png",
|
||
url="/api/files/object?key=collab%2F2026%2F10%2F16%2Fabc.png",
|
||
thumbnail_url="/api/files/object?key=collab%2F2026%2F10%2F16%2Fabc.thumb.webp",
|
||
medium_url="/api/files/object?key=collab%2F2026%2F10%2F16%2Fabc.medium.webp",
|
||
filename="截图.png",
|
||
content_type="image/png",
|
||
size=123,
|
||
)
|
||
assert set(payload.model_dump()) == {
|
||
"object_key",
|
||
"url",
|
||
"thumbnail_url",
|
||
"medium_url",
|
||
"filename",
|
||
"content_type",
|
||
"size",
|
||
}
|
||
|
||
def test_defaults_to_local_when_nothing_is_configured(self, monkeypatch):
|
||
for name in (
|
||
"OSS_MODE",
|
||
"OSS_ENDPOINT",
|
||
"OSS_BUCKET",
|
||
"OSS_REGION",
|
||
"OSS_LOCAL_ROOT",
|
||
"OSS_MAX_UPLOAD_SIZE_MB",
|
||
):
|
||
monkeypatch.delenv(name, raising=False)
|
||
settings = load_storage_settings()
|
||
assert settings.mode == "local"
|
||
assert settings.is_remote is False
|
||
assert settings.max_upload_size_mb == 10
|
||
assert settings.local_root.name == "file_storage"
|
||
|
||
def test_unknown_mode_falls_back_to_local(self, monkeypatch):
|
||
monkeypatch.setenv("OSS_MODE", "s3")
|
||
assert load_storage_settings().mode == "local"
|
||
|
||
def test_region_is_derived_from_internal_endpoint(self, monkeypatch):
|
||
monkeypatch.setenv("OSS_MODE", "oss")
|
||
monkeypatch.setenv(
|
||
"OSS_ENDPOINT", "https://oss-cn-hangzhou-internal.aliyuncs.com"
|
||
)
|
||
monkeypatch.delenv("OSS_REGION", raising=False)
|
||
settings = load_storage_settings()
|
||
assert settings.is_remote is True
|
||
# 生产 .env 只配 endpoint/bucket/AK，region 由端点推导出来。
|
||
assert settings.region == "oss-cn-hangzhou"
|
||
|
||
def test_explicit_region_wins_over_derived_one(self, monkeypatch):
|
||
monkeypatch.setenv("OSS_ENDPOINT", "oss-cn-hangzhou-internal.aliyuncs.com")
|
||
monkeypatch.setenv("OSS_REGION", "oss-cn-shanghai")
|
||
assert load_storage_settings().region == "oss-cn-shanghai"
|
||
|
||
def test_oss_mode_without_endpoint_reports_actionable_error(self, monkeypatch):
|
||
monkeypatch.setenv("OSS_MODE", "oss")
|
||
monkeypatch.delenv("OSS_ENDPOINT", raising=False)
|
||
monkeypatch.delenv("OSS_BUCKET", raising=False)
|
||
storage = get_object_storage(load_storage_settings())
|
||
with pytest.raises(StorageUnavailableError) as excinfo:
|
||
storage.ensure_ready()
|
||
assert "OSS_ENDPOINT" in str(excinfo.value)
|
||
|
||
def test_aliyun_ensure_ready_never_calls_head_bucket(self, monkeypatch):
|
||
"""阿里云桶由控制台创建，HeadBucket 在 RAM 收紧时会 403 甚至挂起，必须跳过。
|
||
|
||
这个分支不构造 client，所以断言「不发任何请求也能返回」。
|
||
"""
|
||
monkeypatch.setenv("OSS_MODE", "oss")
|
||
monkeypatch.setenv(
|
||
"OSS_ENDPOINT", "https://oss-cn-hangzhou-internal.aliyuncs.com"
|
||
)
|
||
monkeypatch.setenv("OSS_BUCKET", "live-hub-assets")
|
||
storage = S3ObjectStorage(load_storage_settings())
|
||
storage.ensure_ready()
|
||
assert storage._client is None
|
||
|
||
def test_s3_client_uses_virtual_host_style_and_derived_region(self, monkeypatch):
|
||
"""真造一个 boto3 client，确认内网端点/region/virtual-host 寻址都生效。"""
|
||
pytest.importorskip("boto3")
|
||
monkeypatch.setenv("OSS_MODE", "oss")
|
||
monkeypatch.setenv(
|
||
"OSS_ENDPOINT", "https://oss-cn-hangzhou-internal.aliyuncs.com"
|
||
)
|
||
monkeypatch.setenv("OSS_BUCKET", "live-hub-assets")
|
||
monkeypatch.setenv("OSS_ACCESS_KEY_ID", "ak")
|
||
monkeypatch.setenv("OSS_ACCESS_KEY_SECRET", "sk")
|
||
storage = S3ObjectStorage(load_storage_settings())
|
||
client = storage._get_client()
|
||
assert client.meta.endpoint_url == (
|
||
"https://oss-cn-hangzhou-internal.aliyuncs.com"
|
||
)
|
||
assert client.meta.region_name == "oss-cn-hangzhou"
|
||
# 阿里云 OSS 只认 virtual-host style。
|
||
assert client.meta.config.s3["addressing_style"] == "virtual"
|
||
|
||
|
||
class TestStoredObjectRead:
|
||
def test_rejects_path_traversal_key(self, local_storage: Path):
|
||
with pytest.raises(FileUploadError) as excinfo:
|
||
get_stored_object("../../etc/passwd", "")
|
||
assert excinfo.value.code == "file_key_invalid"
|
||
|
||
def test_missing_key_raises_not_found(self, local_storage: Path):
|
||
with pytest.raises(FileUploadError) as excinfo:
|
||
get_stored_object("collab/2026/10/16/deadbeef.png", "thumb")
|
||
assert excinfo.value.status_code == 404
|
||
assert excinfo.value.code == "file_not_found"
|
||
|
||
def test_thumbnail_variant_is_served_when_present(self, local_storage: Path):
|
||
session = make_test_session()
|
||
try:
|
||
result = upload_file_asset(
|
||
session,
|
||
data=_png_bytes(1200, 800),
|
||
filename="shot.png",
|
||
content_type="image/png",
|
||
scene="collab",
|
||
)
|
||
finally:
|
||
session.close()
|
||
|
||
thumbnail = get_stored_object(result.object_key, "thumb")
|
||
original = get_stored_object(result.object_key, "")
|
||
try:
|
||
assert thumbnail.content_type == "image/webp"
|
||
assert thumbnail.size < original.size
|
||
assert thumbnail.etag and original.etag
|
||
finally:
|
||
thumbnail.reader.close()
|
||
original.reader.close()
|