"""COS 直传预签名（FR-SYS-03 / T-P1-27）。 全部用 FakeSigner 注入，零网络、零 SDK client 构造。object_key 布局 （场景/用户/日期/uuid）是 COS 侧前缀审计与生命周期规则的基础，逐段精确 断言；场景/类型/大小三重校验共用 40010，逐条锁死。 """ from __future__ import annotations from dataclasses import dataclass, field from datetime import UTC, datetime from typing import Any import pytest from pydantic import ValidationError from packages.shared.config import Settings from packages.shared.errors import AppError, ErrorCode from packages.storage.schemas import PresignRequest from packages.storage.service import StorageService FIXED_NOW = datetime(2026, 8, 14, 1, 2, 3, tzinfo=UTC) @dataclass(frozen=True, slots=True) class FakeSigner: """捕获签发参数并回放可预测 URL，替代 QcloudCosSigner（零网络）。""" calls: list[dict[str, Any]] = field(default_factory=list) def presign_put(self, bucket: str, key: str, expires: int, content_type: str) -> str: self.calls.append( {"bucket": bucket, "key": key, "expires": expires, "content_type": content_type} ) return f"https://cos.fake/{bucket}/{key}?sign=stub" def make_settings(**overrides: Any) -> Settings: """显式 kwargs 构造，测试不受 .env 与宿主环境变量影响。 upload_scenes 定义了 alias 且未开启 populate_by_name，必须按 ``UPLOAD_SCENES_JSON`` 传入——按字段名传会被静默忽略。 """ options: dict[str, Any] = { "_env_file": None, "cos_bucket_temp": "meida-temp-test", "cos_bucket_user": "meida-user-test", "cos_bucket_kb": "meida-kb-test", "cos_bucket_report": "meida-report-test", "cos_presign_expire_seconds": 321, "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) def make_service(**overrides: Any) -> tuple[StorageService, FakeSigner]: signer = FakeSigner() return StorageService(make_settings(**overrides), signer), signer def test_valid_request_builds_auditable_key_layout() -> None: service, signer = make_service() response = service.build_presign( 7, PresignRequest(scene="report_photo", content_type="image/jpeg", size_bytes=100), now=FIXED_NOW, ) # key 布局：{scene}/{user_id}/{yyyyMMdd}/{uuid4hex}.{ext}（写死，改动需过安全评审） prefix = "report_photo/7/20260814/" assert response.object_key.startswith(prefix) filename = response.object_key.removeprefix(prefix) stem, ext = filename.split(".") assert ext == "jpg" assert len(stem) == 32 assert all(char in "0123456789abcdef" for char in stem) # 回显与签发参数 assert ( response.upload_url == f"https://cos.fake/meida-temp-test/{response.object_key}?sign=stub" ) assert response.expires_in == 321 assert response.max_bytes == 1024 assert response.content_type == "image/jpeg" assert signer.calls == [ { "bucket": "meida-temp-test", "key": response.object_key, "expires": 321, "content_type": "image/jpeg", } ] def test_two_presigns_for_the_same_request_never_share_a_key() -> None: service, _ = make_service() request = PresignRequest(scene="avatar", content_type="image/jpeg", size_bytes=1) first = service.build_presign(1, request, now=FIXED_NOW) second = service.build_presign(1, request, now=FIXED_NOW) assert first.object_key != second.object_key # uuid 防猜 @pytest.mark.parametrize( ("body", "message"), [ ( PresignRequest(scene="evil", content_type="image/jpeg", size_bytes=1), "不支持的上传场景", ), (PresignRequest(scene="report_photo", content_type="application/pdf", size_bytes=1), None), (PresignRequest(scene="report_photo", content_type="image/jpeg", size_bytes=1025), None), ], ) def test_scene_type_size_rejections_share_error_code_40010( body: PresignRequest, message: str | None ) -> None: service, signer = make_service() with pytest.raises(AppError) as excinfo: service.build_presign(1, body, now=FIXED_NOW) assert excinfo.value.code is ErrorCode.UPLOAD_REJECTED if message is not None: assert excinfo.value.message == message assert signer.calls == [] # 拒绝路径绝不触发签名 def test_size_equal_to_limit_is_allowed() -> None: service, signer = make_service() service.build_presign( 1, PresignRequest(scene="avatar", content_type="image/webp", size_bytes=1024), now=FIXED_NOW, ) assert signer.calls[0]["key"].endswith(".webp") @pytest.mark.parametrize( ("alias", "expected_bucket"), [ ("temp", "meida-temp-test"), ("user", "meida-user-test"), ("kb", "meida-kb-test"), ("report", "meida-report-test"), ], ) def test_bucket_alias_maps_to_the_matching_settings_field(alias: str, expected_bucket: str) -> None: service, signer = make_service(UPLOAD_SCENES_JSON={"report_photo": {"bucket": alias}}) service.build_presign( 7, PresignRequest(scene="report_photo", content_type="image/png", size_bytes=1), now=FIXED_NOW, ) assert signer.calls[0]["bucket"] == expected_bucket def test_allowed_types_string_tolerates_whitespace() -> None: service, _ = make_service(upload_allowed_types=" image/jpeg , image/png ") response = service.build_presign( 1, PresignRequest(scene="avatar", content_type="image/png", size_bytes=1), now=FIXED_NOW, ) assert response.object_key.endswith(".png") with pytest.raises(AppError) as excinfo: service.build_presign( 1, PresignRequest(scene="avatar", content_type="image/webp", size_bytes=1), now=FIXED_NOW, ) assert excinfo.value.code is ErrorCode.UPLOAD_REJECTED def test_empty_allowed_types_rejects_everything() -> None: service, signer = make_service(upload_allowed_types="") with pytest.raises(AppError) as excinfo: service.build_presign( 1, PresignRequest(scene="avatar", content_type="image/jpeg", size_bytes=1), now=FIXED_NOW, ) assert excinfo.value.code is ErrorCode.UPLOAD_REJECTED assert signer.calls == [] def test_whitelisted_type_without_extension_mapping_is_rejected() -> None: # 白名单被扩到扩展名映射表之外的类型（此处模拟 image/gif）时， # 宁可按 40010 拒绝，也不能在拼 object_key 时抛 KeyError 变成 5xx。 service, signer = make_service(upload_allowed_types="image/gif") with pytest.raises(AppError) as excinfo: service.build_presign( 1, PresignRequest(scene="avatar", content_type="image/gif", size_bytes=1), now=FIXED_NOW, ) assert excinfo.value.code is ErrorCode.UPLOAD_REJECTED assert signer.calls == [] def test_unknown_bucket_alias_is_a_config_error_not_a_client_error() -> None: service, signer = make_service(UPLOAD_SCENES_JSON={"avatar": {"bucket": "ghost"}}) with pytest.raises(AppError) as excinfo: service.build_presign( 1, PresignRequest(scene="avatar", content_type="image/jpeg", size_bytes=1), now=FIXED_NOW, ) assert excinfo.value.code is ErrorCode.CONFIG_ERROR assert signer.calls == [] def test_missing_cos_credentials_is_a_config_error_before_sdk_construction() -> None: # 未注入 signer 且密钥为空：守卫须在 SDK 构造前抛出（零 SDK 接触、零网络）， # 本地未配 COS 四件套时得到结构化 CONFIG_ERROR 而非未处理 500。 service = StorageService(make_settings(), None) with pytest.raises(AppError) as excinfo: service.build_presign( 1, PresignRequest(scene="avatar", content_type="image/jpeg", size_bytes=1), now=FIXED_NOW, ) assert excinfo.value.code is ErrorCode.CONFIG_ERROR def test_request_schema_rejects_non_positive_size() -> None: with pytest.raises(ValidationError): PresignRequest(scene="avatar", content_type="image/jpeg", size_bytes=0)