3868f80502
- 规则 GET 不存在 project → 404（原返误导性空 200，掩盖坏 id）：list_rules 加 项目存在校验（仿 C1/H2）。 - 规则 content 纯空白 → 422（原 strip 前校验 min_length，空白行入库成垃圾）： 改 StringConstraints(strip_whitespace=True, min_length=1)。 - 档位路由 tier 限定 writer/analyst/light（Tier Literal）→ 未知档位 422 （原接受任意字符串）。 回归测试：test_generation（GET rules 404）/ test_rules（空白 content 422）/ test_settings_providers（未知 tier 422）。门禁绿：ruff/format/mypy(210)/pytest。
127 lines
4.5 KiB
Python
127 lines
4.5 KiB
Python
"""T5.5 POST /projects/:id/rules 端点（内存替身，无 DB/无网络）。
|
||
|
||
覆盖：
|
||
- 201 + 回显 level/content + 端点 commit；
|
||
- 非法 level → 422（Pydantic Literal 校验，FastAPI 422）；
|
||
- 空 content → 422；
|
||
- 项目不存在 → 404（QA H2 回归：此前 FK 违例逃逸成 500）。
|
||
"""
|
||
|
||
from __future__ import annotations
|
||
|
||
import uuid
|
||
|
||
import httpx
|
||
import pytest
|
||
from cryptography.fernet import Fernet
|
||
from fakes_projects import FakeProjectRepo, FakeSession
|
||
from ww_api.services.credentials import STUB_OWNER_ID
|
||
from ww_core.domain.project_repo import ProjectView
|
||
from ww_core.domain.rule_repo import RuleWriteView
|
||
|
||
|
||
class _FakeRuleWriteRepo:
|
||
def __init__(self) -> None:
|
||
self.rows: list[RuleWriteView] = []
|
||
|
||
async def create(self, project_id: uuid.UUID, *, level: str, content: str) -> RuleWriteView:
|
||
view = RuleWriteView(project_id=project_id, level=level, content=content)
|
||
self.rows.append(view)
|
||
return view
|
||
|
||
|
||
def _make_client() -> tuple[
|
||
httpx.AsyncClient, _FakeRuleWriteRepo, FakeSession, FakeProjectRepo, uuid.UUID
|
||
]:
|
||
"""构建测试 client，并 seed 一个属于 STUB_OWNER_ID 的项目；返回其 pid。
|
||
|
||
create_rule 现在先校验项目存在（QA H2），故必须 override get_project_repo 并 seed，
|
||
否则正常用例会 404。返回的 pid 是已存在项目；未 seed 的随机 pid 即"不存在"。
|
||
"""
|
||
import os
|
||
|
||
os.environ.setdefault("CREDENTIAL_ENC_KEY", Fernet.generate_key().decode())
|
||
from ww_api.main import create_app
|
||
from ww_api.services.project_deps import get_project_repo, get_rule_write_repo
|
||
from ww_db import get_session
|
||
|
||
repo = _FakeRuleWriteRepo()
|
||
session = FakeSession()
|
||
project_repo = FakeProjectRepo()
|
||
pid = uuid.uuid4()
|
||
project_repo.rows[pid] = (STUB_OWNER_ID, ProjectView(id=pid, title="作品"))
|
||
|
||
app = create_app()
|
||
app.dependency_overrides[get_rule_write_repo] = lambda: repo
|
||
app.dependency_overrides[get_project_repo] = lambda: project_repo
|
||
app.dependency_overrides[get_session] = lambda: session
|
||
transport = httpx.ASGITransport(app=app)
|
||
client = httpx.AsyncClient(transport=transport, base_url="http://test")
|
||
return client, repo, session, project_repo, pid
|
||
|
||
|
||
@pytest.mark.asyncio
|
||
async def test_create_rule_returns_201_and_commits() -> None:
|
||
client, repo, session, _project_repo, pid = _make_client()
|
||
async with client:
|
||
resp = await client.post(
|
||
f"/projects/{pid}/rules",
|
||
json={"level": "project", "content": "主角不许中途复活"},
|
||
)
|
||
assert resp.status_code == 201
|
||
body = resp.json()
|
||
assert body["level"] == "project"
|
||
assert body["content"] == "主角不许中途复活"
|
||
assert session.commits == 1
|
||
assert len(repo.rows) == 1
|
||
assert repo.rows[0].project_id == pid
|
||
|
||
|
||
@pytest.mark.asyncio
|
||
async def test_create_rule_invalid_level_returns_422() -> None:
|
||
client, _repo, _session, _project_repo, pid = _make_client()
|
||
async with client:
|
||
resp = await client.post(
|
||
f"/projects/{pid}/rules",
|
||
json={"level": "nonsense", "content": "x"},
|
||
)
|
||
assert resp.status_code == 422
|
||
|
||
|
||
@pytest.mark.asyncio
|
||
async def test_create_rule_empty_content_returns_422() -> None:
|
||
client, _repo, _session, _project_repo, pid = _make_client()
|
||
async with client:
|
||
resp = await client.post(
|
||
f"/projects/{pid}/rules",
|
||
json={"level": "global", "content": ""},
|
||
)
|
||
assert resp.status_code == 422
|
||
|
||
|
||
@pytest.mark.asyncio
|
||
async def test_create_rule_whitespace_content_returns_422() -> None:
|
||
# QA MEDIUM 回归：纯空白 content（strip 后为空）应 422，不可入库成垃圾行。
|
||
client, repo, _session, _project_repo, pid = _make_client()
|
||
async with client:
|
||
resp = await client.post(
|
||
f"/projects/{pid}/rules",
|
||
json={"level": "project", "content": " "},
|
||
)
|
||
assert resp.status_code == 422
|
||
assert len(repo.rows) == 0
|
||
|
||
|
||
@pytest.mark.asyncio
|
||
async def test_create_rule_unknown_project_returns_404() -> None:
|
||
# QA H2 回归：给不存在的 project 加规则应 404（不是 500 的 FK 违例逃逸）。
|
||
client, repo, _session, _project_repo, _pid = _make_client()
|
||
async with client:
|
||
resp = await client.post(
|
||
f"/projects/{uuid.uuid4()}/rules",
|
||
json={"level": "project", "content": "x"},
|
||
)
|
||
assert resp.status_code == 404
|
||
assert resp.json()["error"]["code"] == "NOT_FOUND"
|
||
assert len(repo.rows) == 0 # 未触达写库
|