meida_backend/tests/unit/test_upload_direct.py

"""本地磁盘直传（无 COS 降级路线）：三重校验 + 落盘布局 + 存在性校验。

零网络零 SDK：落盘用 tmp_path 隔离。require_local_upload 是报告链路挡
模拟/越界 photo_object_key 的关卡，穿越、他人目录、不存在三类逐条锁死。
"""

from __future__ import annotations

from pathlib import Path
from typing import Any

import pytest

from packages.shared.config import Settings
from packages.shared.errors import AppError, ErrorCode
from packages.storage.local import require_local_upload, save_direct_upload

JPEG_BYTES = b"\xff\xd8\xff\xe0" + b"\x00" * 32


def make_settings(tmp_path: Path, **overrides: Any) -> Settings:
    """显式 kwargs 构造（与 test_upload_presign 同口径），测试不受 .env 影响。

    upload_scenes 定义了 alias 且未开启 populate_by_name，必须按
    ``UPLOAD_SCENES_JSON`` 传入——按字段名传会被静默忽略。
    """
    options: dict[str, Any] = {
        "_env_file": None,
        "upload_dir": str(tmp_path / "uploads"),
        "upload_max_bytes": 1024,
        "upload_allowed_types": "image/jpeg,image/png,image/webp",
        "UPLOAD_SCENES_JSON": {"report_photo": {"bucket": "temp"}, "avatar": {"bucket": "temp"}},
    }
    options.update(overrides)
    return Settings(**options)


# --- save_direct_upload -----------------------------------------------------


def test_save_writes_file_and_returns_auditable_key(tmp_path: Path) -> None:
    settings = make_settings(tmp_path)
    result = save_direct_upload(
        settings, user_id=7, scene="report_photo", content_type="image/jpeg", data=JPEG_BYTES
    )

    # key 布局：{scene}/{user_id}/{uuid4hex}.{ext}（与 presign 同构，无日期段——
    # 本地落盘没有 COS 生命周期规则需要按日分组）
    prefix = "report_photo/7/"
    assert result.object_key.startswith(prefix)
    stem, ext = result.object_key.removeprefix(prefix).split(".")
    assert ext == "jpg"
    assert len(stem) == 32
    assert all(char in "0123456789abcdef" for char in stem)

    written = tmp_path / "uploads" / result.object_key
    assert written.read_bytes() == JPEG_BYTES
    assert result.size_bytes == len(JPEG_BYTES)
    assert result.content_type == "image/jpeg"


def test_save_maps_content_type_to_extension(tmp_path: Path) -> None:
    settings = make_settings(tmp_path)
    result = save_direct_upload(
        settings, user_id=1, scene="avatar", content_type="image/webp", data=JPEG_BYTES
    )
    assert result.object_key.endswith(".webp")


def test_save_rejects_unknown_scene(tmp_path: Path) -> None:
    settings = make_settings(tmp_path)
    with pytest.raises(AppError) as excinfo:
        save_direct_upload(
            settings, user_id=1, scene="evil", content_type="image/jpeg", data=JPEG_BYTES
        )
    assert excinfo.value.code is ErrorCode.UPLOAD_REJECTED
    assert excinfo.value.message == "不支持的上传场景"


def test_save_rejects_disallowed_content_type(tmp_path: Path) -> None:
    settings = make_settings(tmp_path)
    with pytest.raises(AppError) as excinfo:
        save_direct_upload(
            settings,
            user_id=1,
            scene="report_photo",
            content_type="application/pdf",
            data=JPEG_BYTES,
        )
    assert excinfo.value.code is ErrorCode.UPLOAD_REJECTED


def test_save_rejects_oversize_and_empty(tmp_path: Path) -> None:
    settings = make_settings(tmp_path)  # upload_max_bytes=1024
    with pytest.raises(AppError) as excinfo:
        save_direct_upload(
            settings,
            user_id=1,
            scene="report_photo",
            content_type="image/jpeg",
            data=b"\x00" * 1025,
        )
    assert excinfo.value.code is ErrorCode.UPLOAD_REJECTED
    with pytest.raises(AppError) as excinfo:
        save_direct_upload(
            settings, user_id=1, scene="report_photo", content_type="image/jpeg", data=b""
        )
    assert excinfo.value.code is ErrorCode.UPLOAD_REJECTED
    # 拒绝路径不落任何文件
    assert not (tmp_path / "uploads").exists()


def test_save_size_equal_to_limit_is_allowed(tmp_path: Path) -> None:
    settings = make_settings(tmp_path)
    result = save_direct_upload(
        settings,
        user_id=1,
        scene="report_photo",
        content_type="image/png",
        data=b"\x00" * 1024,
    )
    assert result.size_bytes == 1024


# --- require_local_upload ---------------------------------------------------


def test_require_roundtrip_returns_existing_file(tmp_path: Path) -> None:
    settings = make_settings(tmp_path)
    saved = save_direct_upload(
        settings, user_id=7, scene="report_photo", content_type="image/jpeg", data=JPEG_BYTES
    )
    path = require_local_upload(
        settings, user_id=7, scene="report_photo", object_key=saved.object_key
    )
    assert path.read_bytes() == JPEG_BYTES


def test_require_rejects_missing_file(tmp_path: Path) -> None:
    settings = make_settings(tmp_path)
    with pytest.raises(AppError) as excinfo:
        require_local_upload(
            settings, user_id=7, scene="report_photo", object_key="report_photo/7/ghost.jpg"
        )
    assert excinfo.value.code is ErrorCode.NOT_FOUND


def test_require_rejects_mock_key_without_upload(tmp_path: Path) -> None:
    # 端侧历史模拟键 report_photo/{userId}/{随机}.jpg：布局合法但无落盘文件。
    settings = make_settings(tmp_path)
    with pytest.raises(AppError) as excinfo:
        require_local_upload(
            settings, user_id=7, scene="report_photo", object_key="report_photo/7/abc123.jpg"
        )
    assert excinfo.value.code is ErrorCode.NOT_FOUND


def test_require_rejects_other_users_key(tmp_path: Path) -> None:
    settings = make_settings(tmp_path)
    saved = save_direct_upload(
        settings, user_id=8, scene="report_photo", content_type="image/jpeg", data=JPEG_BYTES
    )
    with pytest.raises(AppError) as excinfo:
        require_local_upload(settings, user_id=7, scene="report_photo", object_key=saved.object_key)
    assert excinfo.value.code is ErrorCode.NOT_FOUND


@pytest.mark.parametrize(
    "evil_key",
    [
        "report_photo/7/../../avatar/8/x.jpg",  # 路径穿越到他人场景
        "report_photo/7/../../../../etc/passwd",  # 穿越出 upload_dir
        "/etc/passwd",  # 绝对路径吞噬 base
        "avatar/7/x.jpg",  # 场景不匹配
    ],
)
def test_require_rejects_out_of_bounds_keys(tmp_path: Path, evil_key: str) -> None:
    settings = make_settings(tmp_path)
    with pytest.raises(AppError) as excinfo:
        require_local_upload(settings, user_id=7, scene="report_photo", object_key=evil_key)
    assert excinfo.value.code is ErrorCode.NOT_FOUND