dadao
8d61289f90
CI / backend (push) Canceled after 0s
- README：富文本行改成「Web Vditor 编辑 / 安卓 Markwon 预览编辑」，e2e 15 项 - ROADMAP 第 5 节：勾掉表格与超过一两行的旧描述（表格是只读渲染， HTML 片段仍按文本显示）；附件改为短期签名 raw_url；安卓从"未做"改成已完成 - 记下仍欠的三件事：保存后回列表而不是回详情、图片压缩/缩略图、离线缓存 验证：pytest 74 passed；richtext 15 / smoke 7 / tickets 8 / knowledge 7 全绿； tsc -b 0 错；vite build 通过；安卓 assembleDebug 与模拟器逐屏走通。
228 lines
8.9 KiB
Python
228 lines
8.9 KiB
Python
"""附件：上传、列表、下载、删除、跨租户与配额。
|
||
|
||
用例守的是三条约束（见 app/api/attachments.py 文件头）：
|
||
1. 只认本工作区的条目，别人的 item_id 一律 5001；
|
||
2. 超限在读流时卡住，且不留垃圾文件；
|
||
3. 磁盘名服务端生成，用户文件名只进库与 Content-Disposition。
|
||
"""
|
||
|
||
from __future__ import annotations
|
||
|
||
from typing import Any
|
||
|
||
import pytest
|
||
from httpx import AsyncClient
|
||
|
||
import uuid
|
||
|
||
from app.config import get_settings
|
||
from app.services import signing
|
||
|
||
from .conftest import create_workspace, login, unwrap
|
||
|
||
BASE = "/api/v1"
|
||
|
||
|
||
async def _ws_with_item(client: AsyncClient, name: str = "附件工作区") -> tuple[Any, Any, str]:
|
||
user = await login(client)
|
||
ws = await create_workspace(client, user, name)
|
||
item = unwrap(
|
||
await client.post(
|
||
f"{BASE}/workspaces/{ws['id']}/items",
|
||
json={"title": "整理发票原件"},
|
||
headers=user["headers"],
|
||
)
|
||
)
|
||
return user, ws, item["id"]
|
||
|
||
|
||
async def _upload(
|
||
client: AsyncClient, user: dict[str, Any], ws_id: str, item_id: str, content: bytes, name: str
|
||
) -> Any:
|
||
return await client.post(
|
||
f"{BASE}/workspaces/{ws_id}/items/{item_id}/attachments",
|
||
files={"file": (name, content, "text/plain")},
|
||
headers=user["headers"],
|
||
)
|
||
|
||
|
||
async def test_upload_list_and_download(client: AsyncClient) -> None:
|
||
user, ws, item_id = await _ws_with_item(client)
|
||
body = "发票号 044031900111，金额 1280.00".encode()
|
||
|
||
created = unwrap(await _upload(client, user, ws["id"], item_id, body, "发票.txt"))
|
||
assert created["filename"] == "发票.txt"
|
||
assert created["size"] == len(body)
|
||
assert created["sha256"]
|
||
assert created["can_delete"] is True
|
||
|
||
listed = unwrap(
|
||
await client.get(
|
||
f"{BASE}/workspaces/{ws['id']}/items/{item_id}/attachments", headers=user["headers"]
|
||
)
|
||
)
|
||
assert [a["id"] for a in listed["items"]] == [created["id"]]
|
||
assert listed["items"][0]["content_type"] == "text/plain"
|
||
|
||
# 计数要出现在任务列表与详情里，否则前端没法显示「3 个附件」
|
||
items = unwrap(
|
||
await client.get(f"{BASE}/workspaces/{ws['id']}/items?view=all", headers=user["headers"])
|
||
)
|
||
row = next(i for i in items["items"] if i["id"] == item_id)
|
||
assert row["attachment_count"] == 1
|
||
|
||
detail = unwrap(
|
||
await client.get(f"{BASE}/workspaces/{ws['id']}/items/{item_id}", headers=user["headers"])
|
||
)
|
||
assert detail["attachment_count"] == 1
|
||
assert detail["attachments"][0]["id"] == created["id"]
|
||
|
||
# 下载要能原样取回，并给出可辨识的文件名
|
||
resp = await client.get(
|
||
f"{BASE}/workspaces/{ws['id']}/attachments/{created['id']}/download",
|
||
headers=user["headers"],
|
||
)
|
||
assert resp.status_code == 200
|
||
assert resp.content == body
|
||
assert "filename*=UTF-8''" in resp.headers["content-disposition"]
|
||
assert resp.headers["x-content-type-options"] == "nosniff"
|
||
|
||
|
||
async def test_reject_oversized(client: AsyncClient, monkeypatch: pytest.MonkeyPatch) -> None:
|
||
user, ws, item_id = await _ws_with_item(client)
|
||
monkeypatch.setattr(get_settings(), "max_upload_mb", 0)
|
||
|
||
body = (await _upload(client, user, ws["id"], item_id, b"x" * 4096, "big.txt")).json()
|
||
assert body["code"] == 8002
|
||
assert "上限" in body["message"]
|
||
|
||
|
||
async def test_cannot_attach_to_other_tenant_item(client: AsyncClient) -> None:
|
||
"""跨租户的 item_id 不能成为落点——否则等于往别人的时间线上塞文件。"""
|
||
user, _ws, item_id = await _ws_with_item(client, "甲的组")
|
||
other, other_ws, other_item = await _ws_with_item(client, "乙的组")
|
||
|
||
# 拿别人工作区的 id 上传：自己不是成员
|
||
resp = (await _upload(client, user, other_ws["id"], other_item, b"x", "x.txt")).json()
|
||
assert resp["code"] in (3003, 5001)
|
||
|
||
# 在自己的工作区里塞别人的 item_id：条目找不到，也不该暴露它存在
|
||
resp2 = (await _upload(client, user, _ws["id"], other_item, b"x", "x.txt")).json()
|
||
assert resp2["code"] == 5001
|
||
|
||
|
||
async def test_delete_requires_confirm_and_ownership(client: AsyncClient) -> None:
|
||
owner, ws, item_id = await _ws_with_item(client)
|
||
created = unwrap(await _upload(client, owner, ws["id"], item_id, b"abc", "a.txt"))
|
||
|
||
# 不确认不给删：删除不可逆，前端容易误触
|
||
resp = (
|
||
await client.delete(
|
||
f"{BASE}/workspaces/{ws['id']}/attachments/{created['id']}", headers=owner["headers"]
|
||
)
|
||
).json()
|
||
assert resp["code"] == 8004
|
||
assert resp["data"] == {"confirm": True}
|
||
|
||
# 同工作区但非上传者的普通成员也删不了
|
||
outsider = await login(client)
|
||
resp = (
|
||
await client.delete(
|
||
f"{BASE}/workspaces/{ws['id']}/attachments/{created['id']}?confirm=true",
|
||
headers=outsider["headers"],
|
||
)
|
||
).json()
|
||
assert resp["code"] in (3003, 8003)
|
||
|
||
gone = unwrap(
|
||
await client.delete(
|
||
f"{BASE}/workspaces/{ws['id']}/attachments/{created['id']}?confirm=true",
|
||
headers=owner["headers"],
|
||
)
|
||
)
|
||
assert gone["deleted"] is True
|
||
|
||
after = await client.get(
|
||
f"{BASE}/workspaces/{ws['id']}/attachments/{created['id']}/download",
|
||
headers=owner["headers"],
|
||
)
|
||
assert after.json()["code"] == 8001
|
||
|
||
|
||
async def test_raw_image_url_takes_token_in_query(client: AsyncClient) -> None:
|
||
"""Markdown 里的 `` 给 <img> 用：头带不了令牌，只能走 query。"""
|
||
user, ws, item_id = await _ws_with_item(client)
|
||
body = b"\x89PNG\r\n\x1a\n fake png"
|
||
created = unwrap(await _upload(client, user, ws["id"], item_id, body, "图.png"))
|
||
|
||
resp = await client.get(
|
||
f"{BASE}/workspaces/{ws['id']}/attachments/{created['id']}/raw?token={user['token']}"
|
||
)
|
||
assert resp.status_code == 200
|
||
assert resp.content == body
|
||
assert resp.headers["content-disposition"] == "inline"
|
||
assert resp.headers["cache-control"].startswith("private")
|
||
|
||
# 既没签名也没令牌：401（而不是把"缺凭据"说成"文件不存在"）
|
||
missing = await client.get(f"{BASE}/workspaces/{ws['id']}/attachments/{created['id']}/raw")
|
||
assert missing.status_code == 401
|
||
assert missing.json()["code"] == 2001
|
||
outsider = await login(client)
|
||
leaked = await client.get(
|
||
f"{BASE}/workspaces/{ws['id']}/attachments/{created['id']}/raw?token={outsider['token']}"
|
||
)
|
||
assert leaked.json()["code"] == 8001
|
||
|
||
|
||
async def test_raw_url_uses_short_lived_signature(client: AsyncClient) -> None:
|
||
"""图片直链自带短期签名：不需要把登录令牌贴进 URL，也不需要 Authorization 头。"""
|
||
user, ws, item_id = await _ws_with_item(client)
|
||
body = b"\x89PNG\r\n\x1a\n signed png"
|
||
created = unwrap(await _upload(client, user, ws["id"], item_id, body, "签名图.png"))
|
||
|
||
listed = unwrap(
|
||
await client.get(
|
||
f"{BASE}/workspaces/{ws['id']}/items/{item_id}/attachments", headers=user["headers"]
|
||
)
|
||
)
|
||
raw_url = listed["items"][0]["raw_url"]
|
||
assert "exp=" in raw_url and "sig=" in raw_url
|
||
assert "token=" not in raw_url # 不该再把登录令牌塞进 URL
|
||
|
||
# 裸请求（无 Authorization、无 cookie）也要能取到图——这才是 <img> 的真实处境
|
||
resp = await client.get(raw_url)
|
||
assert resp.status_code == 200
|
||
assert resp.content == body
|
||
|
||
|
||
async def test_raw_url_rejects_tampered_or_expired_signature(client: AsyncClient) -> None:
|
||
user, ws, item_id = await _ws_with_item(client)
|
||
created = unwrap(await _upload(client, user, ws["id"], item_id, b"data", "图.png"))
|
||
base = f"{BASE}/workspaces/{ws['id']}/attachments/{created['id']}/raw"
|
||
|
||
# 改一个字符就该失效：签名不是"看起来对就行"
|
||
good = signing.sign(uuid.UUID(ws["id"]), uuid.UUID(created["id"]))
|
||
tampered = good[:-1] + ("0" if good[-1] != "0" else "1")
|
||
bad = await client.get(f"{base}?{tampered}")
|
||
assert bad.status_code == 403
|
||
assert bad.json()["code"] == 8003
|
||
|
||
# 过期签名：就算 HMAC 算得对，时间过了也不给
|
||
stale = signing.sign(uuid.UUID(ws["id"]), uuid.UUID(created["id"]), ttl_seconds=-10)
|
||
expired = await client.get(f"{base}?{stale}")
|
||
assert expired.status_code == 403
|
||
assert expired.json()["code"] == 8003
|
||
|
||
|
||
async def test_raw_url_signature_is_workspace_scoped(client: AsyncClient) -> None:
|
||
"""拿 A 工作区签出来的链接去请求 B 工作区，必须失败（签名里带 workspace_id）。"""
|
||
user, ws_a, item_a = await _ws_with_item(client, "甲工作区")
|
||
ws_b = await create_workspace(client, user, "乙工作区")
|
||
created = unwrap(await _upload(client, user, ws_a["id"], item_a, b"data", "图.png"))
|
||
|
||
query = signing.sign(uuid.UUID(ws_a["id"]), uuid.UUID(created["id"]))
|
||
crossed = await client.get(
|
||
f"{BASE}/workspaces/{ws_b['id']}/attachments/{created['id']}/raw?{query}"
|
||
)
|
||
assert crossed.status_code == 403
|
||
assert crossed.json()["code"] == 8003 |