fix(设置): 恢复上传提交失败前的公司标志文件

数据库提交失败时恢复被覆盖的文件,或移除本次新文件。
文件恢复失败保留原始异常,并补充三种扩展名场景的回归测试。
This commit is contained in:
liuwenxin 2026-11-03 17:18:39 +08:00
parent 67adecbf1b
commit c684a44753
2 changed files with 46 additions and 7 deletions

View File

@ -126,14 +126,23 @@ def save_logo(
setting = get_setting(session) setting = get_setting(session)
old_path = setting.logo_path old_path = setting.logo_path
# 先写新文件、commit 成功后再清旧扩展名(若把清理放在 commit 之前, # 保留同名旧文件内容,避免覆盖后数据库提交失败造成 logo 断链;
# commit 失败会让库指旧路径而旧文件已删——断链无补偿);同扩展名 # 不同扩展名的新文件在提交失败时删除,不改变当前生效文件。
# 由写入覆盖,无需也不会误删新文件
target = _logo_dir() / f"logo.{ext}" target = _logo_dir() / f"logo.{ext}"
target.write_bytes(content) previous_content = target.read_bytes() if target.exists() else None
try:
setting.logo_path = f"logo/logo.{ext}" target.write_bytes(content)
session.commit() setting.logo_path = f"logo/logo.{ext}"
session.commit()
except Exception:
try:
if previous_content is None:
target.unlink(missing_ok=True)
else:
target.write_bytes(previous_content)
except OSError:
logger.warning("logo 数据库回滚后的文件恢复失败", exc_info=True)
raise
old_ext = Path(old_path).suffix.lstrip(".") if old_path else None old_ext = Path(old_path).suffix.lstrip(".") if old_path else None
if old_ext and old_ext != ext: if old_ext and old_ext != ext:

View File

@ -0,0 +1,30 @@
"""数据库提交失败时,上传文件恢复到原状且保留原始异常。"""
from contextlib import nullcontext
from types import SimpleNamespace
from unittest.mock import Mock
import pytest
from app.services import system_setting_service as service
@pytest.mark.parametrize("old_ext", [None, "png", "jpg"])
def test_提交失败不丢旧logo也不残留新文件(tmp_path, monkeypatch, old_ext):
original = b"original-logo"
if old_ext:
(tmp_path / f"logo.{old_ext}").write_bytes(original)
monkeypatch.setattr(service, "_logo_dir", lambda: tmp_path)
setting = SimpleNamespace(logo_path=f"logo/logo.{old_ext}" if old_ext else None)
monkeypatch.setattr(service, "get_setting", lambda _: setting)
monkeypatch.setattr(service.audit_service, "audit", lambda *_, **__: nullcontext())
session = Mock()
session.commit.side_effect = RuntimeError("数据库提交失败")
with pytest.raises(RuntimeError, match="数据库提交失败"):
service.save_logo(session, service._MAGIC["png"] + b"new", "png", operator=Mock())
if old_ext == "png":
assert (tmp_path / "logo.png").read_bytes() == original
else:
assert not (tmp_path / "logo.png").exists()
if old_ext:
assert (tmp_path / f"logo.{old_ext}").read_bytes() == original