331 lines
13 KiB
Python
331 lines
13 KiB
Python
"""系统设置（公司信息）接口测试。
|
||
|
||
守的是：
|
||
- 单行语义：全系统恒一行，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
|
||
_ICO_BYTES = b"\x00\x00\x01\x00" + b"\x00" * 32
|
||
_TXT_BYTES = b"not an image at all"
|
||
|
||
|
||
@pytest.fixture
|
||
def client() -> TestClient:
|
||
return TestClient(app)
|
||
|
||
|
||
_ASSET_KINDS = ("logo", "favicon")
|
||
|
||
|
||
def _snapshot_assets() -> dict[str, dict[str, bytes]]:
|
||
"""快照上传目录里的真实图片（logo/favicon）。"""
|
||
from app.core.config import get_settings
|
||
|
||
root = get_settings().upload_dir
|
||
snapshot: dict[str, dict[str, bytes]] = {}
|
||
for kind in _ASSET_KINDS:
|
||
directory = root / kind
|
||
snapshot[kind] = (
|
||
{path.name: path.read_bytes() for path in directory.glob("*") if path.is_file()}
|
||
if directory.is_dir()
|
||
else {}
|
||
)
|
||
return snapshot
|
||
|
||
|
||
def _restore_assets(snapshot: dict[str, dict[str, bytes]]) -> None:
|
||
"""还原上传目录：删除测试新增文件、回写被测试覆盖的原文件。
|
||
|
||
上传用例会用固定文件名覆盖 logo/favicon，并清理旧扩展名文件 —— 只还原
|
||
数据库路径而不还原文件，会把开发环境里真实上传的 logo 永久删掉。
|
||
"""
|
||
from app.core.config import get_settings
|
||
|
||
root = get_settings().upload_dir
|
||
for kind, files in snapshot.items():
|
||
directory = root / kind
|
||
directory.mkdir(parents=True, exist_ok=True)
|
||
for path in directory.glob("*"):
|
||
if path.is_file() and path.name not in files:
|
||
path.unlink(missing_ok=True)
|
||
for name, content in files.items():
|
||
(directory / name).write_bytes(content)
|
||
|
||
|
||
@pytest.fixture(autouse=True)
|
||
def _restore_company_info():
|
||
"""系统设置是**真实单行数据 + 真实上传文件**：用例前快照、结束后还原。
|
||
|
||
与其他表的测试数据不同，公司信息没有 t_ 前缀可供兜底清理 —— 不还原
|
||
的话，跑完测试后开发环境的侧边栏品牌/登录页/浏览器标题会被改成
|
||
「测试公司_xxx」，上传用例还会覆盖/删除真实 logo 与站点图标。
|
||
"""
|
||
from app.core.db import get_session_factory
|
||
from app.models import SystemSetting
|
||
|
||
factory = get_session_factory()
|
||
with factory() as session:
|
||
row = session.get(SystemSetting, 1)
|
||
snapshot = (
|
||
{column.name: getattr(row, column.name) for column in SystemSetting.__table__.columns}
|
||
if row is not None
|
||
else None
|
||
)
|
||
asset_snapshot = _snapshot_assets()
|
||
|
||
yield
|
||
|
||
# 文件先还原：数据库路径随后回指同一批文件
|
||
_restore_assets(asset_snapshot)
|
||
if snapshot is None:
|
||
return
|
||
with factory() as session:
|
||
row = session.get(SystemSetting, 1)
|
||
if row is None:
|
||
return
|
||
for key, value in snapshot.items():
|
||
if key != "update_time": # 更新时间由数据库 ON UPDATE 维护
|
||
setattr(row, key, value)
|
||
session.commit()
|
||
|
||
|
||
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",
|
||
"site_title",
|
||
"brand_slogan",
|
||
"footer_text",
|
||
"icp_no",
|
||
"favicon_url",
|
||
}, "public 端点字段越界"
|
||
assert data["company_name"] == name
|
||
assert resp.headers.get("cache-control") is not None
|
||
|
||
def test_已删除的档案字段不再返回(self, client: TestClient) -> None:
|
||
"""V8 精简：副标题/信用代码/地址/电话/邮箱已从模型与契约中删除。"""
|
||
data = client.get(f"{_SETTINGS}/public").json()["data"]
|
||
for field in (
|
||
"system_subtitle",
|
||
"credit_code",
|
||
"company_address",
|
||
"contact_phone",
|
||
"contact_email",
|
||
):
|
||
assert field not in data, f"已删除字段仍出现在匿名端点：{field}"
|
||
|
||
def test_写操作匿名不可用(self, client: TestClient) -> None:
|
||
assert client.put(f"{_SETTINGS}/public", json={"company_name": "x"}).status_code in {
|
||
401,
|
||
405,
|
||
}
|
||
|
||
|
||
class TestCompanyProfile:
|
||
"""站点标题/品牌信息（V8）：部分更新语义与空串清空。"""
|
||
|
||
def test_部分更新_未提交字段保持不变(self, client: TestClient, admin: dict) -> None:
|
||
name = f"站点公司_{uuid.uuid4().hex[:6]}"
|
||
client.put(_SETTINGS, json={"company_name": name}, headers=admin["headers"])
|
||
|
||
put = client.put(_SETTINGS, json={"site_title": "测试站点标题"}, headers=admin["headers"])
|
||
assert put.status_code == 200, put.text
|
||
assert put.json()["data"]["site_title"] == "测试站点标题"
|
||
# 未提交的 company_name 保持不变（部分更新语义）
|
||
assert put.json()["data"]["company_name"] == name
|
||
|
||
got = client.get(_SETTINGS, headers=admin["headers"]).json()["data"]
|
||
assert got["company_name"] == name
|
||
assert got["site_title"] == "测试站点标题"
|
||
|
||
def test_可选文本传空串_清空为null(self, client: TestClient, admin: dict) -> None:
|
||
client.put(
|
||
_SETTINGS,
|
||
json={"brand_slogan": "t_临时标语", "icp_no": "t_备案号"},
|
||
headers=admin["headers"],
|
||
)
|
||
put = client.put(
|
||
_SETTINGS, json={"brand_slogan": "", "icp_no": ""}, headers=admin["headers"]
|
||
)
|
||
assert put.status_code == 200, put.text
|
||
assert put.json()["data"]["brand_slogan"] is None
|
||
assert put.json()["data"]["icp_no"] is None
|
||
|
||
def test_超长字段被拒(self, client: TestClient, admin: dict) -> None:
|
||
resp = client.put(_SETTINGS, json={"site_title": "x" * 65}, headers=admin["headers"])
|
||
assert resp.status_code == 400, resp.text
|
||
|
||
|
||
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
|
||
|
||
def test_同扩展名替换_url版本号变化(self, client: TestClient, admin: dict) -> None:
|
||
"""同扩展名覆盖时路径与 update_time 都不变 —— 版本号必须递增，否则浏览器用旧图。"""
|
||
first = self._upload(client, admin, "logo.png", _PNG_BYTES).json()["data"]["logo_url"]
|
||
second = self._upload(client, admin, "logo.png", _PNG_BYTES).json()["data"]["logo_url"]
|
||
assert first and second
|
||
assert first != second, "同扩展名替换后图片 URL 未变化，浏览器会继续使用旧缓存"
|
||
|
||
|
||
class TestFaviconUpload:
|
||
"""站点图标上传校验链（独立于 logo；favicon 额外接受 ico）。"""
|
||
|
||
def _upload(self, client: TestClient, admin: dict, filename: str, content: bytes):
|
||
return client.post(
|
||
f"{_SETTINGS}/favicon",
|
||
files={"file": (filename, content, "application/octet-stream")},
|
||
headers=admin["headers"],
|
||
)
|
||
|
||
def test_合法ico上传成功且回读(self, client: TestClient, admin: dict) -> None:
|
||
resp = self._upload(client, admin, "favicon.ico", _ICO_BYTES)
|
||
assert resp.status_code == 200, resp.text
|
||
out = resp.json()["data"]
|
||
assert out["favicon_url"] and "/favicon/public" in out["favicon_url"]
|
||
|
||
img = client.get(f"{_SETTINGS}/favicon/public")
|
||
assert img.status_code == 200
|
||
assert img.headers["content-type"] == "image/x-icon"
|
||
|
||
def test_合法png上传成功(self, client: TestClient, admin: dict) -> None:
|
||
resp = self._upload(client, admin, "favicon.png", _PNG_BYTES)
|
||
assert resp.status_code == 200, resp.text
|
||
assert "/favicon/public" in resp.json()["data"]["favicon_url"]
|
||
|
||
def test_伪造扩展名被magic_bytes拦截(self, client: TestClient, admin: dict) -> None:
|
||
resp = self._upload(client, admin, "fake.ico", _TXT_BYTES)
|
||
assert resp.status_code == 400, resp.text
|
||
assert "不符" in resp.json()["message"]
|
||
|
||
def test_同扩展名替换_url版本号变化(self, client: TestClient, admin: dict) -> None:
|
||
first = self._upload(client, admin, "favicon.png", _PNG_BYTES).json()["data"]["favicon_url"]
|
||
second = self._upload(client, admin, "favicon.png", _PNG_BYTES).json()["data"][
|
||
"favicon_url"
|
||
]
|
||
assert first and second
|
||
assert first != second, "同扩展名替换后站点图标 URL 未变化"
|
||
|
||
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}/favicon",
|
||
files={"file": ("favicon.png", _PNG_BYTES, "application/octet-stream")},
|
||
)
|
||
assert resp.status_code == 401
|