MemeMeow/tests/test_image_upload_http.py

"""公共图片上传 HTTP 边界契约测试。"""

from __future__ import annotations

import ast
import asyncio
from pathlib import Path
from types import SimpleNamespace

import pytest
from fastapi import HTTPException

import api
from backend import image_upload_http
from backend.operation_policy import OperationPolicyError


class _Upload:
    """提供受控文件名和异步读取行为的 multipart 文件替身。"""

    def __init__(self, filename: str, content: bytes) -> None:
        """保存单文件输入和读取游标。"""
        self.filename = filename
        self._content = content
        self._read = False

    async def read(self, size: int) -> bytes:
        """按受限读取接口返回内容，第二次读取返回 EOF。"""
        if self._read:
            return b""
        self._read = True
        return self._content[:size]


class _Metadata:
    """记录 scope-bound metadata 访问和 durable 上传副作用。"""

    def __init__(self, root: Path, events: list[str]) -> None:
        """初始化测试文件根目录及事件记录。"""
        self.root = root
        self.events = events
        self.blob_store = SimpleNamespace(resolve=self.resolve)

    def resolve(self, key: str, *, must_exist: bool = False) -> Path:
        """返回当前 scope 根目录下的受控目标。"""
        self.events.append("resolve")
        return self.root / key

    def find_existing_upload(self, key: str, *, sha256: str, size_bytes: int):
        """默认表示目标尚未存在。"""
        self.events.append("find_existing_upload")
        return None

    def upload_bytes(self, content: bytes, *, target_key: str):
        """写入测试文件并返回旧 metadata service 形状。"""
        self.events.append("upload_bytes")
        path = self.root / target_key
        path.write_bytes(content)
        return "meme-1", path

    def status(self, image: Path) -> dict[str, str]:
        """返回稳定 metadata 状态。"""
        self.events.append("status")
        return {"status": "pending"}


def _request(tmp_path: Path, events: list[str], files: list[_Upload], *, fields: set[str] | None = None) -> SimpleNamespace:
    """构造只满足上传 callback 所需字段的最小 Request 替身。"""
    settings = SimpleNamespace(max_files_per_request=20, max_request_bytes=None, max_upload_size=1024)
    state = SimpleNamespace(settings=settings)
    request = SimpleNamespace(app=SimpleNamespace(state=state), fields=fields or {"files"}, events=events)
    request.form = SimpleNamespace(keys=lambda: request.fields, get=lambda key: None, getlist=lambda key: files if key == "files" else [])
    return request


def _error(status: int, code: str, message: str) -> HTTPException:
    """构造与入口相同 detail 形状的测试异常。"""
    return HTTPException(status_code=status, detail={"error": code, "message": message})


def _options(*_args, **_kwargs) -> SimpleNamespace:
    """返回默认图片处理选项。"""
    return SimpleNamespace(reverse_image_policy="forbid", auto_name=False)


def _common_kwargs(request: SimpleNamespace, metadata: _Metadata, events: list[str], tmp_path: Path) -> dict[str, object]:
    """为模块函数注入最小、显式且 scope-bound 的依赖。"""
    request.form = SimpleNamespace(keys=lambda: request.fields, get=lambda key: None, getlist=lambda key: request._files if key == "files" else [])
    return {
        "settings": lambda received: received.app.state.settings,
        "metadata_service": lambda received: metadata,
        "task_service": lambda received: SimpleNamespace(seal_batch=lambda _batch: events.append("seal"), schedule=lambda _task: events.append("schedule")),
        "normalize_processing_options": _options,
        "parse_multipart_bool": lambda value, default=False: default if value is None else bool(value),
        "sanitize_filename": lambda value: value,
        "validate_image": lambda content, suffix: events.append("validate_image"),
        "calculate_sha256": lambda content: "a" * 64,
        "processing_worker": lambda received: None,
        "submit_processing_job": lambda *args, **kwargs: (_ for _ in ()).throw(RuntimeError("no_worker")),
        "processing_config": lambda received: {},
        "submit_visual_task": lambda *args, **kwargs: SimpleNamespace(task_id="task-1", status="pending"),
        "context_enqueue_error": lambda exc: "enqueue_failed",
        "acquire_operation": lambda *args, **kwargs: (events.append("acquire") or "grant"),
        "commit_operation": lambda received, grant: events.append("commit"),
        "release_operation": lambda received, grant: events.append("release"),
        "invalidate_search": lambda received: events.append("invalidate"),
        "error": _error,
    }


def test_upload_route_and_module_dependency_are_single_directional() -> None:
    """上传 canonical route 只注册一次，公共模块不反向依赖入口。"""
    matching = [route for route in api.app.routes if route.path == "/images/upload"]
    assert len(matching) == 1
    assert matching[0].methods == {"POST"}
    assert matching[0].tags == ["images"]
    source = Path(image_upload_http.__file__).read_text(encoding="utf-8")
    tree = ast.parse(source)
    imported = {node.module.split(".", 1)[0] for node in ast.walk(tree) if isinstance(node, ast.ImportFrom) and node.module}
    imported.update(alias.name.split(".", 1)[0] for node in ast.walk(tree) if isinstance(node, ast.Import) for alias in node.names)
    assert "api" not in imported
    assert "server_api" not in imported
    assert api.upload_images.__name__ == "upload_images"
    assert api._parse_upload_form is image_upload_http._parse_upload_form
    assert api._read_upload_content is image_upload_http._read_upload_content


def test_upload_rejects_unknown_field_before_scope_service(tmp_path: Path, monkeypatch: pytest.MonkeyPatch) -> None:
    """未知 multipart 字段在任何 metadata 或文件副作用前被拒绝。"""
    events: list[str] = []
    request = _request(tmp_path, events, [_Upload("a.png", b"png")], fields={"files", "directory"})
    request._files = [_Upload("a.png", b"png")]
    metadata = _Metadata(tmp_path, events)
    async def parse_form(*_args, **_kwargs):
        return request.form
    monkeypatch.setattr(image_upload_http, "_parse_upload_form", parse_form)
    with pytest.raises(HTTPException) as caught:
        asyncio.run(image_upload_http.upload_images(request, **_common_kwargs(request, metadata, events, tmp_path)))
    assert caught.value.status_code == 400
    assert caught.value.detail["error"] == "invalid_request"
    assert events == []


def test_upload_commits_before_processing_and_search_invalidation(tmp_path: Path, monkeypatch: pytest.MonkeyPatch) -> None:
    """新图片必须按 acquire、durable、commit、任务和检索失效顺序收束。"""
    events: list[str] = []
    request = _request(tmp_path, events, [_Upload("a.png", b"png")])
    request._files = [_Upload("a.png", b"png")]
    metadata = _Metadata(tmp_path, events)
    async def parse_form(*_args, **_kwargs):
        return request.form
    monkeypatch.setattr(image_upload_http, "_parse_upload_form", parse_form)
    result = asyncio.run(image_upload_http.upload_images(request, **_common_kwargs(request, metadata, events, tmp_path)))
    assert result["results"][0]["ok"] is True
    assert events.index("find_existing_upload") < events.index("acquire") < events.index("upload_bytes") < events.index("commit")
    assert events.index("status") < events.index("invalidate")


def test_upload_policy_rejection_is_fail_closed(tmp_path: Path, monkeypatch: pytest.MonkeyPatch) -> None:
    """policy 拒绝不得触碰 durable upload。"""
    events: list[str] = []
    request = _request(tmp_path, events, [_Upload("a.png", b"png")])
    request._files = [_Upload("a.png", b"png")]
    metadata = _Metadata(tmp_path, events)
    async def parse_form(*_args, **_kwargs):
        return request.form
    monkeypatch.setattr(image_upload_http, "_parse_upload_form", parse_form)
    kwargs = _common_kwargs(request, metadata, events, tmp_path)
    kwargs["acquire_operation"] = lambda *args, **kwargs: (_ for _ in ()).throw(OperationPolicyError("operation_forbidden"))
    kwargs["operation_error"] = lambda exc: _error(403, exc.code, "forbidden")
    with pytest.raises(HTTPException) as caught:
        asyncio.run(image_upload_http.upload_images(request, **kwargs))
    assert caught.value.status_code == 403
    assert "upload_bytes" not in events
    assert "commit" not in events


def test_public_upload_filename_omits_hash_fallback_without_display_record() -> None:
    """旧 facade 只返回 meme_id 时，上传结果不能把内容寻址 key 当文件名返回。"""
    digest = "a" * 64
    record = SimpleNamespace(sha256=digest, extension=".png")

    assert image_upload_http._public_saved_filename(SimpleNamespace(), record, f"{digest}.png") is None
    assert image_upload_http._public_saved_filename(SimpleNamespace(), "meme-1", f"{digest}.png") is None
    assert image_upload_http._public_saved_filename(SimpleNamespace(), "meme-1", "原名.png") == "原名.png"