feat(settings): 公司信息配置与 logo 管理(public 端点供登录页)
- system_settings 单行 CRUD:GET/PUT(AdminUser)+ GET /public(无鉴权, SettingPublicOut 白名单仅 company_name/logo_url 两字段,短缓存 300s) - logo 上传:后缀+magic bytes+2MB 三重校验,固定名覆盖落盘 uploads/logo/ (不使用上传方文件名,杜绝路径拼接);旧扩展名文件清理;文件流端点 no-cache,URL 带 update_time 版本参数破缓存 - logo 存目录不存库:图片入 DB 让每次读取与 mysqldump 放大传输且无法 独立替换;DB 只持一行相对路径 - 新增依赖 python-multipart(FastAPI UploadFile 必备传输层);.gitignore 增 backend/uploads/ 与 db/backups/ - 集成 11 例:单行语义/白名单字段/三重校验/越权 403/匿名 401/审计留痕 Co-Authored-By: Claude Code <noreply@anthropic.com>
This commit is contained in:
parent
e95dc4b2de
commit
efed29647c
5
.gitignore
vendored
5
.gitignore
vendored
@ -59,3 +59,8 @@ frontend/e2e/*.png
|
|||||||
backup/
|
backup/
|
||||||
.claude
|
.claude
|
||||||
CLAUDE.md
|
CLAUDE.md
|
||||||
|
|
||||||
|
# 运行期上传物(公司 logo 等):与构建产物隔离,不入版本库
|
||||||
|
backend/uploads/
|
||||||
|
# 数据库备份导出目录(⑨ 备份脚本产物)
|
||||||
|
db/backups/
|
||||||
|
|||||||
@ -19,6 +19,7 @@ from app.api.v1.system import business_roles as system_business_roles
|
|||||||
from app.api.v1.system import logs as system_logs
|
from app.api.v1.system import logs as system_logs
|
||||||
from app.api.v1.system import params as system_params
|
from app.api.v1.system import params as system_params
|
||||||
from app.api.v1.system import roles as system_roles
|
from app.api.v1.system import roles as system_roles
|
||||||
|
from app.api.v1.system import settings as system_settings
|
||||||
from app.api.v1.system import users as system_users
|
from app.api.v1.system import users as system_users
|
||||||
|
|
||||||
api_router = APIRouter()
|
api_router = APIRouter()
|
||||||
@ -36,6 +37,7 @@ system_router.include_router(system_users.router)
|
|||||||
system_router.include_router(system_roles.router)
|
system_router.include_router(system_roles.router)
|
||||||
system_router.include_router(system_business_roles.router)
|
system_router.include_router(system_business_roles.router)
|
||||||
system_router.include_router(system_params.router)
|
system_router.include_router(system_params.router)
|
||||||
|
system_router.include_router(system_settings.router)
|
||||||
system_router.include_router(system_logs.operation_router)
|
system_router.include_router(system_logs.operation_router)
|
||||||
system_router.include_router(system_logs.login_router)
|
system_router.include_router(system_logs.login_router)
|
||||||
v1_router.include_router(system_router)
|
v1_router.include_router(system_router)
|
||||||
|
|||||||
75
backend/app/api/v1/system/settings.py
Normal file
75
backend/app/api/v1/system/settings.py
Normal file
@ -0,0 +1,75 @@
|
|||||||
|
"""系统设置接口(公司信息)。
|
||||||
|
|
||||||
|
`GET /public` 与 `GET /logo/public` **无鉴权**:登录页渲染公司名/logo 时
|
||||||
|
尚无令牌。响应经 `SettingPublicOut` 白名单收敛(见 schemas/setting.py),
|
||||||
|
仅暴露两个非敏感展示字段。
|
||||||
|
|
||||||
|
`/logo/public` 返回文件流而非统一信封——与参数导出同一例外口径
|
||||||
|
(二进制响应套 JSON 信封无意义)。
|
||||||
|
"""
|
||||||
|
|
||||||
|
from fastapi import APIRouter, Response, UploadFile
|
||||||
|
from fastapi.responses import FileResponse
|
||||||
|
|
||||||
|
from app.api.deps import AdminUser, DbSession, RequestMetaDep
|
||||||
|
from app.core.response import ApiResponse, ok
|
||||||
|
from app.schemas.setting import SettingOut, SettingPublicOut, SettingUpdateIn
|
||||||
|
from app.services import system_setting_service as service
|
||||||
|
|
||||||
|
router = APIRouter(prefix="/settings", tags=["系统管理 · 公司信息"])
|
||||||
|
|
||||||
|
_LOGO_MEDIA_TYPE = {"png": "image/png", "jpg": "image/jpeg"}
|
||||||
|
|
||||||
|
|
||||||
|
@router.get("", response_model=ApiResponse[SettingOut], summary="公司信息")
|
||||||
|
def get_settings(session: DbSession, _: AdminUser) -> ApiResponse[SettingOut]:
|
||||||
|
"""取公司信息(管理端)。"""
|
||||||
|
return ok(service.to_out(service.get_setting(session)))
|
||||||
|
|
||||||
|
|
||||||
|
@router.put("", response_model=ApiResponse[SettingOut], summary="更新公司名称")
|
||||||
|
def update_settings(
|
||||||
|
payload: SettingUpdateIn,
|
||||||
|
session: DbSession,
|
||||||
|
current_user: AdminUser,
|
||||||
|
meta: RequestMetaDep,
|
||||||
|
) -> ApiResponse[SettingOut]:
|
||||||
|
"""更新公司名称(审计留痕)。"""
|
||||||
|
return ok(
|
||||||
|
service.update_company(session, payload, operator=current_user, meta=meta),
|
||||||
|
message="已保存",
|
||||||
|
)
|
||||||
|
|
||||||
|
|
||||||
|
@router.post("/logo", response_model=ApiResponse[SettingOut], summary="上传 logo")
|
||||||
|
async def upload_logo(
|
||||||
|
file: UploadFile,
|
||||||
|
session: DbSession,
|
||||||
|
current_user: AdminUser,
|
||||||
|
meta: RequestMetaDep,
|
||||||
|
) -> ApiResponse[SettingOut]:
|
||||||
|
"""上传公司 logo(png/jpg ≤2MB;固定名覆盖,旧扩展名文件清理)。"""
|
||||||
|
content = await file.read()
|
||||||
|
ext = (file.filename or "").rsplit(".", 1)[-1]
|
||||||
|
return ok(
|
||||||
|
service.save_logo(session, content, ext, operator=current_user, meta=meta),
|
||||||
|
message="logo 已更新",
|
||||||
|
)
|
||||||
|
|
||||||
|
|
||||||
|
@router.get("/public", response_model=ApiResponse[SettingPublicOut], summary="公司信息(匿名)")
|
||||||
|
def get_settings_public(session: DbSession, response: Response) -> ApiResponse[SettingPublicOut]:
|
||||||
|
"""公司信息(登录页消费,无鉴权)。短缓存减轻匿名端点压力。"""
|
||||||
|
response.headers["Cache-Control"] = "public, max-age=300"
|
||||||
|
return ok(service.to_public_out(service.get_setting(session)))
|
||||||
|
|
||||||
|
|
||||||
|
@router.get("/logo/public", summary="logo 图片(匿名)")
|
||||||
|
def get_logo_public(session: DbSession) -> FileResponse:
|
||||||
|
"""logo 文件流(无鉴权)。no-cache:图片替换后由版本参数破缓存兜底。"""
|
||||||
|
path, ext = service.read_logo(session)
|
||||||
|
return FileResponse(
|
||||||
|
path,
|
||||||
|
media_type=_LOGO_MEDIA_TYPE.get(ext, "application/octet-stream"),
|
||||||
|
headers={"Cache-Control": "no-cache"},
|
||||||
|
)
|
||||||
@ -92,6 +92,15 @@ class Settings(BaseSettings):
|
|||||||
auth = f":{self.redis_password}@" if self.redis_password else ""
|
auth = f":{self.redis_password}@" if self.redis_password else ""
|
||||||
return f"redis://{auth}{self.redis_host}:{self.redis_port}/{self.redis_db}"
|
return f"redis://{auth}{self.redis_host}:{self.redis_port}/{self.redis_db}"
|
||||||
|
|
||||||
|
@property
|
||||||
|
def upload_dir(self) -> Path:
|
||||||
|
"""上传文件根目录(公司 logo 等):backend/uploads/。
|
||||||
|
|
||||||
|
不叫 static/ 是刻意的:静态目录将来可能整体交给前端构建产物,
|
||||||
|
运行期上传物必须与之隔离;该目录不入版本库(.gitignore)。
|
||||||
|
"""
|
||||||
|
return BACKEND_DIR / "uploads"
|
||||||
|
|
||||||
|
|
||||||
@lru_cache
|
@lru_cache
|
||||||
def get_settings() -> Settings:
|
def get_settings() -> Settings:
|
||||||
|
|||||||
28
backend/app/schemas/setting.py
Normal file
28
backend/app/schemas/setting.py
Normal file
@ -0,0 +1,28 @@
|
|||||||
|
"""系统设置(公司信息)请求/响应模型。
|
||||||
|
|
||||||
|
`SettingPublicOut` 与 `SettingOut` **刻意不共用**:public 端点无鉴权,
|
||||||
|
响应字段必须白名单显式列出——将来往 SettingOut 加敏感字段时不会
|
||||||
|
顺着复用链泄漏到匿名端点。
|
||||||
|
"""
|
||||||
|
|
||||||
|
from pydantic import BaseModel, Field
|
||||||
|
|
||||||
|
|
||||||
|
class SettingOut(BaseModel):
|
||||||
|
"""公司信息(管理端)。"""
|
||||||
|
|
||||||
|
company_name: str
|
||||||
|
logo_url: str | None = None
|
||||||
|
|
||||||
|
|
||||||
|
class SettingPublicOut(BaseModel):
|
||||||
|
"""公司信息(匿名端点白名单):登录页展示所需的两个字段。"""
|
||||||
|
|
||||||
|
company_name: str
|
||||||
|
logo_url: str | None = None
|
||||||
|
|
||||||
|
|
||||||
|
class SettingUpdateIn(BaseModel):
|
||||||
|
"""公司信息更新。"""
|
||||||
|
|
||||||
|
company_name: str = Field(min_length=1, max_length=64, description="公司名称")
|
||||||
165
backend/app/services/system_setting_service.py
Normal file
165
backend/app/services/system_setting_service.py
Normal file
@ -0,0 +1,165 @@
|
|||||||
|
"""系统设置(公司信息)业务编排。
|
||||||
|
|
||||||
|
单例行语义:整个系统恒一行(id=1,迁移种子保底)。logo 只存**相对路径**
|
||||||
|
(uploads/logo/ 下固定名覆盖),文件不入库的理由:图片进数据库会让每次
|
||||||
|
读取与备份放大传输,且无法独立替换;DB 只持有一行指针。
|
||||||
|
|
||||||
|
logo 安全:后缀白名单 + magic bytes 双重校验(上传内容不可信,仅看文件名
|
||||||
|
后缀可被伪造);固定文件名落盘,**不使用上传方文件名**——杜绝路径拼接与
|
||||||
|
覆盖任意文件。
|
||||||
|
"""
|
||||||
|
|
||||||
|
import logging
|
||||||
|
from pathlib import Path
|
||||||
|
|
||||||
|
from sqlalchemy.orm import Session
|
||||||
|
|
||||||
|
from app.core.config import get_settings
|
||||||
|
from app.core.constants import ACTION_UPDATE, MODULE_SYSTEM_SETTINGS
|
||||||
|
from app.core.exceptions import BadRequestError, NotFoundError
|
||||||
|
from app.models import SystemSetting, User
|
||||||
|
from app.schemas.setting import SettingOut, SettingPublicOut, SettingUpdateIn
|
||||||
|
from app.services import audit_service
|
||||||
|
from app.services.audit_service import AuditRecord, RequestMeta
|
||||||
|
|
||||||
|
logger = logging.getLogger(__name__)
|
||||||
|
|
||||||
|
# magic bytes:PNG 固定 8 字节签名;JPEG 以 SOI + APP0/JFIF 起始
|
||||||
|
_MAGIC = {
|
||||||
|
"png": b"\x89PNG\r\n\x1a\n",
|
||||||
|
"jpg": b"\xff\xd8\xff",
|
||||||
|
}
|
||||||
|
LOGO_MAX_BYTES = 2 * 1024 * 1024 # 2MB:页眉/登录页展示尺寸上限,防上传滥用
|
||||||
|
|
||||||
|
SETTINGS_ID = 1 # 全系统单行主键
|
||||||
|
|
||||||
|
|
||||||
|
def get_setting(session: Session) -> SystemSetting:
|
||||||
|
"""取公司信息单行(种子保底 id=1;防御式兜底:极端环境下缺行则补建)。"""
|
||||||
|
setting = session.get(SystemSetting, SETTINGS_ID)
|
||||||
|
if setting is None:
|
||||||
|
setting = SystemSetting(id=SETTINGS_ID, company_name="衍水集团")
|
||||||
|
session.add(setting)
|
||||||
|
session.flush()
|
||||||
|
return setting
|
||||||
|
|
||||||
|
|
||||||
|
def to_out(setting: SystemSetting) -> SettingOut:
|
||||||
|
"""管理端出参(logo_url 带更新时间版本参数,替换图片后浏览器不再用旧缓存)。"""
|
||||||
|
return SettingOut(
|
||||||
|
company_name=setting.company_name,
|
||||||
|
logo_url=_logo_url(setting),
|
||||||
|
)
|
||||||
|
|
||||||
|
|
||||||
|
def to_public_out(setting: SystemSetting) -> SettingPublicOut:
|
||||||
|
"""匿名端点出参(白名单字段,独立 schema 见模块说明)。"""
|
||||||
|
return SettingPublicOut(
|
||||||
|
company_name=setting.company_name,
|
||||||
|
logo_url=_logo_url(setting),
|
||||||
|
)
|
||||||
|
|
||||||
|
|
||||||
|
def _logo_url(setting: SystemSetting) -> str | None:
|
||||||
|
if not setting.logo_path:
|
||||||
|
return None
|
||||||
|
version = int(setting.update_time.timestamp()) if setting.update_time else 0
|
||||||
|
return f"/api/v1/system/settings/logo/public?v={version}"
|
||||||
|
|
||||||
|
|
||||||
|
def update_company(
|
||||||
|
session: Session,
|
||||||
|
payload: SettingUpdateIn,
|
||||||
|
*,
|
||||||
|
operator: User,
|
||||||
|
meta: RequestMeta | None = None,
|
||||||
|
) -> SettingOut:
|
||||||
|
"""更新公司名称(审计留痕)。"""
|
||||||
|
with audit_service.audit(
|
||||||
|
session,
|
||||||
|
operator=operator,
|
||||||
|
record=AuditRecord(
|
||||||
|
module=MODULE_SYSTEM_SETTINGS,
|
||||||
|
action=ACTION_UPDATE,
|
||||||
|
summary=f"修改公司名称为「{payload.company_name}」",
|
||||||
|
request_data=payload.model_dump(),
|
||||||
|
),
|
||||||
|
meta=meta,
|
||||||
|
):
|
||||||
|
setting = get_setting(session)
|
||||||
|
setting.company_name = payload.company_name.strip()
|
||||||
|
session.commit()
|
||||||
|
return to_out(setting)
|
||||||
|
|
||||||
|
|
||||||
|
def save_logo(
|
||||||
|
session: Session,
|
||||||
|
content: bytes,
|
||||||
|
ext: str,
|
||||||
|
*,
|
||||||
|
operator: User,
|
||||||
|
meta: RequestMeta | None = None,
|
||||||
|
) -> SettingOut:
|
||||||
|
"""上传 logo:校验 → 固定名覆盖落盘 → 更新相对路径(审计留痕)。
|
||||||
|
|
||||||
|
旧扩展名文件一并清理(logo.png 换 jpg 后残留文件会继续占盘)。
|
||||||
|
"""
|
||||||
|
ext = ext.lower()
|
||||||
|
if ext not in _MAGIC:
|
||||||
|
raise BadRequestError("logo 仅支持 png/jpg 格式")
|
||||||
|
if len(content) > LOGO_MAX_BYTES:
|
||||||
|
raise BadRequestError("logo 文件不能超过 2MB")
|
||||||
|
if not content.startswith(_MAGIC[ext]):
|
||||||
|
raise BadRequestError("文件内容与扩展名不符(仅接受真实图片)")
|
||||||
|
|
||||||
|
with audit_service.audit(
|
||||||
|
session,
|
||||||
|
operator=operator,
|
||||||
|
record=AuditRecord(
|
||||||
|
module=MODULE_SYSTEM_SETTINGS,
|
||||||
|
action=ACTION_UPDATE,
|
||||||
|
summary="更新公司 logo",
|
||||||
|
request_data={"ext": ext, "size": len(content)},
|
||||||
|
),
|
||||||
|
meta=meta,
|
||||||
|
):
|
||||||
|
setting = get_setting(session)
|
||||||
|
old_path = setting.logo_path
|
||||||
|
|
||||||
|
# 旧扩展名清理必须在写新文件**之前**:同扩展名覆盖时先写后删会把
|
||||||
|
# 刚落盘的新文件一并删掉(回读 404)
|
||||||
|
old_ext = Path(old_path).suffix.lstrip(".") if old_path else None
|
||||||
|
if old_ext and old_ext != ext:
|
||||||
|
_cleanup_old_logo(old_ext)
|
||||||
|
|
||||||
|
target = _logo_dir() / f"logo.{ext}"
|
||||||
|
target.write_bytes(content)
|
||||||
|
|
||||||
|
setting.logo_path = f"logo/logo.{ext}"
|
||||||
|
session.commit()
|
||||||
|
logger.info("更新公司 logo | size=%s ext=%s", len(content), ext)
|
||||||
|
return to_out(setting)
|
||||||
|
|
||||||
|
|
||||||
|
def read_logo(session: Session) -> tuple[Path, str]:
|
||||||
|
"""按库中记录定位 logo 文件,返回 (路径, 扩展名)。未上传抛 404。"""
|
||||||
|
setting = get_setting(session)
|
||||||
|
if not setting.logo_path:
|
||||||
|
raise NotFoundError("尚未上传 logo")
|
||||||
|
path = get_settings().upload_dir / setting.logo_path
|
||||||
|
if not path.is_file():
|
||||||
|
raise NotFoundError("logo 文件不存在,请重新上传")
|
||||||
|
return path, path.suffix.lstrip(".")
|
||||||
|
|
||||||
|
|
||||||
|
def _logo_dir() -> Path:
|
||||||
|
directory = get_settings().upload_dir / "logo"
|
||||||
|
directory.mkdir(parents=True, exist_ok=True)
|
||||||
|
return directory
|
||||||
|
|
||||||
|
|
||||||
|
def _cleanup_old_logo(old_ext: str) -> None:
|
||||||
|
if old_ext in _MAGIC:
|
||||||
|
old_file = _logo_dir() / f"logo.{old_ext}"
|
||||||
|
if old_file.exists():
|
||||||
|
old_file.unlink(missing_ok=True)
|
||||||
@ -35,6 +35,7 @@ dependencies = [
|
|||||||
"captcha>=0.6",
|
"captcha>=0.6",
|
||||||
"pillow>=11.0",
|
"pillow>=11.0",
|
||||||
"openpyxl>=3.1.5",
|
"openpyxl>=3.1.5",
|
||||||
|
"python-multipart>=0.0.32",
|
||||||
]
|
]
|
||||||
|
|
||||||
[dependency-groups]
|
[dependency-groups]
|
||||||
|
|||||||
137
backend/tests/integration/test_settings.py
Normal file
137
backend/tests/integration/test_settings.py
Normal file
@ -0,0 +1,137 @@
|
|||||||
|
"""系统设置(公司信息)接口测试。
|
||||||
|
|
||||||
|
守的是:
|
||||||
|
- 单行语义:全系统恒一行,PUT 只改名称不增行
|
||||||
|
- public 端点无鉴权可用但**白名单字段**——多一个敏感字段都是匿名泄漏
|
||||||
|
- logo 上传三重校验(后缀/magic bytes/大小)——仅看后缀可被伪造
|
||||||
|
- 审计留痕:系统设置属敏感配置,改必须有日志
|
||||||
|
|
||||||
|
真图片构造:PNG 用 8 字节签名 + 最小 IHDR;JPG 用 SOI 头 + 填充字节。
|
||||||
|
仅校验 magic bytes 前缀,无需完整合法图片(解析由前端 img 完成)。
|
||||||
|
"""
|
||||||
|
|
||||||
|
import uuid
|
||||||
|
|
||||||
|
import pytest
|
||||||
|
from fastapi.testclient import TestClient
|
||||||
|
|
||||||
|
from app.main import app
|
||||||
|
|
||||||
|
pytestmark = pytest.mark.integration
|
||||||
|
|
||||||
|
_SETTINGS = "/api/v1/system/settings"
|
||||||
|
|
||||||
|
_PNG_BYTES = b"\x89PNG\r\n\x1a\n" + b"\x00" * 32
|
||||||
|
_JPG_BYTES = b"\xff\xd8\xff\xe0" + b"\x00" * 32
|
||||||
|
_TXT_BYTES = b"not an image at all"
|
||||||
|
|
||||||
|
|
||||||
|
@pytest.fixture
|
||||||
|
def client() -> TestClient:
|
||||||
|
return TestClient(app)
|
||||||
|
|
||||||
|
|
||||||
|
class TestCompanyInfo:
|
||||||
|
"""公司名称读取与更新。"""
|
||||||
|
|
||||||
|
def test_初始值与更新生效(self, client: TestClient, admin: dict) -> None:
|
||||||
|
got = client.get(_SETTINGS, headers=admin["headers"]).json()["data"]
|
||||||
|
assert got["company_name"]
|
||||||
|
# logo_url 允许为 None(未上传)或历史遗留的 URL——单行库可复用,
|
||||||
|
# 断言不依赖「干净库」前提;上传行为由 TestLogoUpload 专类覆盖
|
||||||
|
assert got["logo_url"] is None or "/logo/public" in got["logo_url"]
|
||||||
|
|
||||||
|
new_name = f"测试公司_{uuid.uuid4().hex[:6]}"
|
||||||
|
put = client.put(_SETTINGS, json={"company_name": new_name}, headers=admin["headers"])
|
||||||
|
assert put.status_code == 200, put.text
|
||||||
|
assert put.json()["data"]["company_name"] == new_name
|
||||||
|
assert (
|
||||||
|
client.get(_SETTINGS, headers=admin["headers"]).json()["data"]["company_name"]
|
||||||
|
== new_name
|
||||||
|
)
|
||||||
|
|
||||||
|
def test_名称为空被拒(self, client: TestClient, admin: dict) -> None:
|
||||||
|
assert (
|
||||||
|
client.put(_SETTINGS, json={"company_name": ""}, headers=admin["headers"]).status_code
|
||||||
|
== 400
|
||||||
|
)
|
||||||
|
|
||||||
|
def test_非管理员改名称被拒(self, client: TestClient, normal: dict) -> None:
|
||||||
|
resp = client.put(_SETTINGS, json={"company_name": "越权公司"}, headers=normal["headers"])
|
||||||
|
assert resp.status_code == 403
|
||||||
|
|
||||||
|
def test_修改留审计日志(self, client: TestClient, admin: dict) -> None:
|
||||||
|
client.put(
|
||||||
|
_SETTINGS,
|
||||||
|
json={"company_name": f"审计公司_{uuid.uuid4().hex[:6]}"},
|
||||||
|
headers=admin["headers"],
|
||||||
|
)
|
||||||
|
logs = client.get(
|
||||||
|
"/api/v1/system/operation-logs",
|
||||||
|
params={"module": "系统设置"},
|
||||||
|
headers=admin["headers"],
|
||||||
|
).json()["data"]["list"]
|
||||||
|
assert logs, "修改公司名称未产生操作日志"
|
||||||
|
|
||||||
|
|
||||||
|
class TestPublicEndpoint:
|
||||||
|
"""匿名端点:白名单字段 + 可用性。"""
|
||||||
|
|
||||||
|
def test_无令牌可读且仅白名单字段(self, client: TestClient, admin: dict) -> None:
|
||||||
|
name = f"公开公司_{uuid.uuid4().hex[:6]}"
|
||||||
|
client.put(_SETTINGS, json={"company_name": name}, headers=admin["headers"])
|
||||||
|
|
||||||
|
resp = client.get(f"{_SETTINGS}/public")
|
||||||
|
assert resp.status_code == 200
|
||||||
|
data = resp.json()["data"]
|
||||||
|
assert set(data) == {"company_name", "logo_url"}, "public 端点字段越界"
|
||||||
|
assert data["company_name"] == name
|
||||||
|
assert resp.headers.get("cache-control") is not None
|
||||||
|
|
||||||
|
def test_写操作匿名不可用(self, client: TestClient) -> None:
|
||||||
|
assert client.put(f"{_SETTINGS}/public", json={"company_name": "x"}).status_code in {
|
||||||
|
401,
|
||||||
|
405,
|
||||||
|
}
|
||||||
|
|
||||||
|
|
||||||
|
class TestLogoUpload:
|
||||||
|
"""logo 上传校验链。"""
|
||||||
|
|
||||||
|
def _upload(self, client: TestClient, admin: dict, filename: str, content: bytes):
|
||||||
|
return client.post(
|
||||||
|
f"{_SETTINGS}/logo",
|
||||||
|
files={"file": (filename, content, "application/octet-stream")},
|
||||||
|
headers=admin["headers"],
|
||||||
|
)
|
||||||
|
|
||||||
|
def test_合法png上传成功且回读(self, client: TestClient, admin: dict) -> None:
|
||||||
|
resp = self._upload(client, admin, "logo.png", _PNG_BYTES)
|
||||||
|
assert resp.status_code == 200, resp.text
|
||||||
|
out = resp.json()["data"]
|
||||||
|
assert out["logo_url"] and "/logo/public" in out["logo_url"]
|
||||||
|
|
||||||
|
img = client.get(f"{_SETTINGS}/logo/public")
|
||||||
|
assert img.status_code == 200
|
||||||
|
assert img.headers["content-type"] == "image/png"
|
||||||
|
|
||||||
|
def test_伪造扩展名被magic_bytes拦截(self, client: TestClient, admin: dict) -> None:
|
||||||
|
resp = self._upload(client, admin, "fake.png", _TXT_BYTES)
|
||||||
|
assert resp.status_code == 400
|
||||||
|
assert "不符" in resp.json()["message"]
|
||||||
|
|
||||||
|
def test_非法格式被拒(self, client: TestClient, admin: dict) -> None:
|
||||||
|
assert self._upload(client, admin, "x.gif", _PNG_BYTES).status_code == 400
|
||||||
|
|
||||||
|
def test_超限被拒(self, client: TestClient, admin: dict) -> None:
|
||||||
|
big = _PNG_BYTES + b"\x00" * (2 * 1024 * 1024)
|
||||||
|
resp = self._upload(client, admin, "big.png", big)
|
||||||
|
assert resp.status_code == 400
|
||||||
|
assert "2MB" in resp.json()["message"]
|
||||||
|
|
||||||
|
def test_匿名上传被拒(self, client: TestClient) -> None:
|
||||||
|
resp = client.post(
|
||||||
|
f"{_SETTINGS}/logo",
|
||||||
|
files={"file": ("logo.png", _PNG_BYTES, "application/octet-stream")},
|
||||||
|
)
|
||||||
|
assert resp.status_code == 401
|
||||||
11
backend/uv.lock
generated
11
backend/uv.lock
generated
@ -435,6 +435,7 @@ dependencies = [
|
|||||||
{ name = "pydantic-settings" },
|
{ name = "pydantic-settings" },
|
||||||
{ name = "pyjwt" },
|
{ name = "pyjwt" },
|
||||||
{ name = "pymysql", extra = ["rsa"] },
|
{ name = "pymysql", extra = ["rsa"] },
|
||||||
|
{ name = "python-multipart" },
|
||||||
{ name = "redis", extra = ["hiredis"] },
|
{ name = "redis", extra = ["hiredis"] },
|
||||||
{ name = "sqlalchemy" },
|
{ name = "sqlalchemy" },
|
||||||
{ name = "starlette" },
|
{ name = "starlette" },
|
||||||
@ -462,6 +463,7 @@ requires-dist = [
|
|||||||
{ name = "pydantic-settings", specifier = ">=2.6" },
|
{ name = "pydantic-settings", specifier = ">=2.6" },
|
||||||
{ name = "pyjwt", specifier = ">=2.10" },
|
{ name = "pyjwt", specifier = ">=2.10" },
|
||||||
{ name = "pymysql", extras = ["rsa"], specifier = ">=1.2.3" },
|
{ name = "pymysql", extras = ["rsa"], specifier = ">=1.2.3" },
|
||||||
|
{ name = "python-multipart", specifier = ">=0.0.32" },
|
||||||
{ name = "redis", extras = ["hiredis"], specifier = ">=5.2" },
|
{ name = "redis", extras = ["hiredis"], specifier = ">=5.2" },
|
||||||
{ name = "sqlalchemy", specifier = ">=2.0.36" },
|
{ name = "sqlalchemy", specifier = ">=2.0.36" },
|
||||||
{ name = "starlette", specifier = ">=1.7,<2" },
|
{ name = "starlette", specifier = ">=1.7,<2" },
|
||||||
@ -656,6 +658,15 @@ wheels = [
|
|||||||
{ url = "https://pypi.tuna.tsinghua.edu.cn/packages/60/d1/38f3a3405989a89ac18390803e70c6ad7c7760da4f9b83cbeca0c44a0c72/python_dotenv-1.2.4-py3-none-any.whl", hash = "sha256:42269a8a5b3fd54ffa6f3d84b18abed50064717576b4ecf03dc4a55d8aa04fdc", size = 23266, upload-time = "2026-10-01T05:36:08.633Z" },
|
{ url = "https://pypi.tuna.tsinghua.edu.cn/packages/60/d1/38f3a3405989a89ac18390803e70c6ad7c7760da4f9b83cbeca0c44a0c72/python_dotenv-1.2.4-py3-none-any.whl", hash = "sha256:42269a8a5b3fd54ffa6f3d84b18abed50064717576b4ecf03dc4a55d8aa04fdc", size = 23266, upload-time = "2026-10-01T05:36:08.633Z" },
|
||||||
]
|
]
|
||||||
|
|
||||||
|
[[package]]
|
||||||
|
name = "python-multipart"
|
||||||
|
version = "0.0.32"
|
||||||
|
source = { registry = "https://pypi.tuna.tsinghua.edu.cn/simple" }
|
||||||
|
sdist = { url = "https://pypi.tuna.tsinghua.edu.cn/packages/5b/42/55c32bb9b12693c092ad250a0e82edb5b31ddeda6eb772de5f308b3804ad/python_multipart-0.0.32.tar.gz", hash = "sha256:be54b7f3fa167bb83e4fcd936b887b708f4e57fe75911c02aebf53efaf8d938e", size = 46881, upload-time = "2026-06-04T16:18:58.647Z" }
|
||||||
|
wheels = [
|
||||||
|
{ url = "https://pypi.tuna.tsinghua.edu.cn/packages/e1/04/e8135ebd1ad02c56ec633277529b2602ff99ff634be76cdba5744cf554fd/python_multipart-0.0.32-py3-none-any.whl", hash = "sha256:ff6d3f776f16878c894e52e107296ffc890e913c611b1a4ec6c44e2821fe2e23", size = 30042, upload-time = "2026-06-04T16:18:57.319Z" },
|
||||||
|
]
|
||||||
|
|
||||||
[[package]]
|
[[package]]
|
||||||
name = "pyyaml"
|
name = "pyyaml"
|
||||||
version = "6.0.3"
|
version = "6.0.3"
|
||||||
|
|||||||
Loading…
Reference in New Issue
Block a user