fix(m0): 迁移可执行 + 测试全绿(33 条)
CI / backend (push) Canceled after 0s

跑通迁移与测试后修掉的真问题:

- alembic 0001:asyncpg 走预备语句,一次 execute 只能一条 SQL。
  CREATE FUNCTION / CREATE POLICY / GRANT 全部拆成单语句逐条执行,
  否则报 "cannot insert multiple commands into a prepared statement"。
- 创建工作区:ORM flush 只按 relationship 排序,裸外键不算数,
  member 可能先于 workspace 落地撞 FK。改为先 flush workspace 再插 member。
- datetime.UTC 是 3.11+ 才有的,本机 3.10 直接 ImportError。改用 timezone.utc。
- 幂等中间件:原来会缓存 4xx。校验失败被缓存后,用户改对参数仍拿到同一个 422。
  改为只缓存 <400。
- 发码节流窗口改为可配置 SMS_THROTTLE_SECONDS(测试置 0),
  节流用例自己显式打开,避免"整库关掉节流"导致节流从未被验证。
- 租户隔离用例:pytest.raises(Exception) 改为精确的 ProgrammingError,
  否则任何异常都会让"应用角色不能建表"这条断言假通过。

验收:pytest 33 passed;ruff check / format 通过;真机跑 uvicorn,
/health、/version、登录、建工作区、列表、/auth/me 全通。
This commit is contained in:
Steven
2026-10-28 03:22:59 +08:00
parent 0c53aa7062
commit 2b493bd9f8
14 changed files with 160 additions and 136 deletions
+16 -23
View File
@@ -13,6 +13,7 @@ import uuid
import pytest
from httpx import AsyncClient
from sqlalchemy import text
from sqlalchemy.exc import ProgrammingError
from app.db import SessionFactory, bind_tenant, current_app_role
from app.models import TENANT_TABLES
@@ -37,9 +38,7 @@ async def test_cannot_read_other_workspace(client: AsyncClient) -> None:
async def test_cannot_list_other_workspace_members(client: AsyncClient) -> None:
alice, _bob, _ws_a, ws_b = await _two_tenants(client)
resp = await client.get(
f"/api/v1/workspaces/{ws_b['id']}/members", headers=alice["headers"]
)
resp = await client.get(f"/api/v1/workspaces/{ws_b['id']}/members", headers=alice["headers"])
assert resp.status_code == 403
@@ -66,15 +65,12 @@ async def test_rls_hides_rows_at_database_level(client: AsyncClient) -> None:
ws_a_id = uuid.UUID(ws_a["id"])
ws_b_id = uuid.UUID(ws_b["id"])
async with SessionFactory() as session:
async with session.begin():
await bind_tenant(session, ws_a_id, None)
visible = (
await session.execute(text("SELECT id FROM workspace"))
).scalars().all()
member_ids = (
await session.execute(text("SELECT workspace_id FROM member"))
).scalars().all()
async with SessionFactory() as session, session.begin():
await bind_tenant(session, ws_a_id, None)
visible = (await session.execute(text("SELECT id FROM workspace"))).scalars().all()
member_ids = (
(await session.execute(text("SELECT workspace_id FROM member"))).scalars().all()
)
assert [str(v) for v in visible] == [ws_a["id"]]
assert set(member_ids) == {ws_a_id}
@@ -83,9 +79,8 @@ async def test_rls_hides_rows_at_database_level(client: AsyncClient) -> None:
async def test_rls_without_context_returns_nothing(client: AsyncClient) -> None:
await _two_tenants(client)
async with SessionFactory() as session:
async with session.begin():
count = (await session.execute(text("SELECT count(*) FROM workspace"))).scalar_one()
async with SessionFactory() as session, session.begin():
count = (await session.execute(text("SELECT count(*) FROM workspace"))).scalar_one()
assert count == 0, "未绑定租户上下文时不应看到任何行"
@@ -100,15 +95,14 @@ async def test_rls_enabled_on_every_tenant_table(client: AsyncClient) -> None:
{"names": list(TENANT_TABLES)},
)
).all()
status = {name: enabled for name, enabled in rows}
status = dict(rows)
assert set(status) == set(TENANT_TABLES), "有租户表没建出来"
assert all(status.values()), f"未启用 RLS 的表:{[k for k, v in status.items() if not v]}"
async def test_app_role_cannot_bypass_rls(client: AsyncClient) -> None:
async with SessionFactory() as session:
async with session.begin():
role = await current_app_role(session)
async with SessionFactory() as session, session.begin():
role = await current_app_role(session)
assert role["is_super"] is False, "应用角色绝不能是 superuser,否则 RLS 被跳过"
assert role["bypass_rls"] is False, "应用角色不能带 BYPASSRLS"
assert role["read_all"] is False
@@ -116,7 +110,6 @@ async def test_app_role_cannot_bypass_rls(client: AsyncClient) -> None:
async def test_app_role_has_no_ddl(client: AsyncClient) -> None:
"""应用角色不能建表——否则一次 SQL 注入就能改结构。"""
async with SessionFactory() as session:
async with session.begin():
with pytest.raises(Exception):
await session.execute(text("CREATE TABLE should_not_exist (id int)"))
async with SessionFactory() as session, session.begin():
with pytest.raises(ProgrammingError): # 权限不足,DDL 必须失败
await session.execute(text("CREATE TABLE should_not_exist (id int)"))