"""在线协作图片上传：存储层 + 服务层的行为契约。 只跑本地目录后端（``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()