完成错误测试用例
This commit is contained in:
Your Name
@@ -0,0 +1,223 @@
|
||||
"""
|
||||
项目管理系统 API（依据产品文档与 API 文档）
|
||||
TDD：路由按测试用例实现
|
||||
"""
|
||||
import uuid
|
||||
from datetime import datetime
|
||||
from typing import Any, Optional
|
||||
|
||||
from fastapi import FastAPI, Header, HTTPException
|
||||
from fastapi.middleware.cors import CORSMiddleware
|
||||
from fastapi.responses import JSONResponse
|
||||
from pydantic import BaseModel
|
||||
|
||||
app = FastAPI(title="项目管理系统 API", version="1.0.0")
|
||||
app.add_middleware(CORSMiddleware, allow_origins=["*"], allow_methods=["*"], allow_headers=["*"])
|
||||
|
||||
# ---------- 内存存储（TDD 绿阶段最小实现，后续可换 DB）----------
|
||||
# token -> role；测试用：valid-token / market-token -> 市场部, engineer-token -> 工程部
|
||||
TOKEN_ROLE: dict[str, str] = {
|
||||
"valid-token": "市场部",
|
||||
"market-token": "市场部",
|
||||
"engineer-token": "工程部",
|
||||
}
|
||||
# 合法用户：仅 market / valid-password 登录成功
|
||||
VALID_USER = {"username": "market", "password": "valid-password", "role": "市场部", "display_name": "市场部用户"}
|
||||
# 项目 id -> 项目数据（含 contract, costControl, receivable, payable, other）
|
||||
PROJECTS: dict[str, dict[str, Any]] = {}
|
||||
# 操作日志 project_id -> list of log
|
||||
OPERATION_LOGS: dict[str, list[dict]] = {}
|
||||
|
||||
# 不预置项目，测试通过 create 或 fixture 造数；列表无数据时返回 []
|
||||
|
||||
|
||||
# ---------- 请求/响应模型（最小）----------
|
||||
class LoginBody(BaseModel):
|
||||
username: Optional[str] = None
|
||||
password: Optional[str] = None
|
||||
|
||||
|
||||
class ProjectListBody(BaseModel):
|
||||
page: Optional[int] = 1
|
||||
pageSize: Optional[int] = 20
|
||||
searchType: Optional[str] = None
|
||||
keyword: Optional[str] = None
|
||||
progress: Optional[str] = None
|
||||
cost: Optional[str] = None
|
||||
|
||||
|
||||
class ProjectDetailBody(BaseModel):
|
||||
id: str
|
||||
|
||||
|
||||
class ContractCreate(BaseModel):
|
||||
contractCode: Optional[str] = None
|
||||
projectName: Optional[str] = None
|
||||
|
||||
|
||||
class ProjectCreateBody(BaseModel):
|
||||
contract: Optional[ContractCreate] = None
|
||||
costControl: Optional[dict] = None
|
||||
receivable: Optional[dict] = None
|
||||
payable: Optional[dict] = None
|
||||
other: Optional[dict] = None
|
||||
|
||||
|
||||
class ProjectUpdateBody(BaseModel):
|
||||
id: str
|
||||
contract: Optional[dict] = None
|
||||
costControl: Optional[dict] = None
|
||||
receivable: Optional[dict] = None
|
||||
payable: Optional[dict] = None
|
||||
other: Optional[dict] = None
|
||||
|
||||
|
||||
class ProjectLogsBody(BaseModel):
|
||||
id: str
|
||||
page: Optional[int] = 1
|
||||
pageSize: Optional[int] = 20
|
||||
|
||||
|
||||
def _get_role(authorization: Optional[str] = Header(None)) -> str:
|
||||
if not authorization or not authorization.startswith("Bearer "):
|
||||
raise HTTPException(status_code=401, detail="未登录")
|
||||
token = authorization[7:].strip()
|
||||
role = TOKEN_ROLE.get(token)
|
||||
if role is None:
|
||||
raise HTTPException(status_code=401, detail="未登录")
|
||||
return role
|
||||
|
||||
|
||||
def _get_token(authorization: Optional[str] = Header(None)) -> str:
|
||||
if not authorization or not authorization.startswith("Bearer "):
|
||||
raise HTTPException(status_code=401, detail="未登录")
|
||||
return authorization[7:].strip()
|
||||
|
||||
|
||||
# ---------- 认证 ----------
|
||||
@app.post("/api/auth/login")
|
||||
def login(body: LoginBody):
|
||||
if body.username is None or (isinstance(body.username, str) and body.username.strip() == ""):
|
||||
raise HTTPException(status_code=400, detail="缺少 username")
|
||||
if body.password is None or (isinstance(body.password, str) and body.password.strip() == ""):
|
||||
raise HTTPException(status_code=400, detail="缺少 password")
|
||||
if body.username != VALID_USER["username"] or body.password != VALID_USER["password"]:
|
||||
return JSONResponse(status_code=401, content={"code": 401, "message": "账号或密码错误"})
|
||||
token = "valid-token"
|
||||
return {
|
||||
"token": token,
|
||||
"user": {
|
||||
"id": "user-market",
|
||||
"username": VALID_USER["username"],
|
||||
"role": VALID_USER["role"],
|
||||
"displayName": VALID_USER["display_name"],
|
||||
},
|
||||
}
|
||||
|
||||
|
||||
# ---------- 项目列表 ----------
|
||||
@app.post("/api/projects/list")
|
||||
def project_list(body: ProjectListBody, authorization: Optional[str] = Header(None)):
|
||||
_get_role(authorization)
|
||||
page = body.page or 1
|
||||
page_size = body.pageSize or 20
|
||||
items = list(PROJECTS.values())
|
||||
# 列表项：id, projectName, contractCode, progress, cost, updatedAt
|
||||
list_ = [
|
||||
{
|
||||
"id": p["id"],
|
||||
"projectName": (p.get("contract") or {}).get("projectName") or "",
|
||||
"contractCode": (p.get("contract") or {}).get("contractCode") or "",
|
||||
"progress": p.get("progress") or "",
|
||||
"cost": p.get("cost") or "",
|
||||
"updatedAt": p.get("updatedAt") or "",
|
||||
}
|
||||
for p in items
|
||||
]
|
||||
return {"list": list_, "total": len(list_), "page": page, "pageSize": page_size}
|
||||
|
||||
|
||||
# ---------- 项目详情 ----------
|
||||
@app.post("/api/projects/detail")
|
||||
def project_detail(body: ProjectDetailBody, authorization: Optional[str] = Header(None)):
|
||||
_get_role(authorization)
|
||||
pid = body.id
|
||||
if pid not in PROJECTS:
|
||||
return JSONResponse(status_code=404, content={"code": 404, "message": "项目不存在"})
|
||||
p = PROJECTS[pid]
|
||||
return {
|
||||
"id": p["id"],
|
||||
"contract": p.get("contract") or {},
|
||||
"costControl": p.get("costControl") or {},
|
||||
"receivable": p.get("receivable") or {},
|
||||
"payable": p.get("payable") or {},
|
||||
"other": p.get("other") or {},
|
||||
"createdAt": p.get("createdAt", ""),
|
||||
"updatedAt": p.get("updatedAt", ""),
|
||||
}
|
||||
|
||||
|
||||
# ---------- 新建项目 ----------
|
||||
@app.post("/api/projects/create", status_code=201)
|
||||
def project_create(body: ProjectCreateBody, authorization: Optional[str] = Header(None)):
|
||||
role = _get_role(authorization)
|
||||
if role != "市场部":
|
||||
return JSONResponse(status_code=403, content={"code": 403, "message": "无权限"})
|
||||
contract = body.contract
|
||||
if contract is None:
|
||||
return JSONResponse(status_code=400, content={"code": 400, "message": "合同编号不能为空"})
|
||||
contract_code = getattr(contract, "contractCode", None)
|
||||
project_name = getattr(contract, "projectName", None)
|
||||
if contract_code is None or (isinstance(contract_code, str) and contract_code.strip() == ""):
|
||||
return JSONResponse(status_code=400, content={"code": 400, "message": "合同编号不能为空"})
|
||||
if project_name is None or (isinstance(project_name, str) and project_name.strip() == ""):
|
||||
return JSONResponse(status_code=400, content={"code": 400, "message": "项目名称不能为空"})
|
||||
pid = str(uuid.uuid4())
|
||||
now = datetime.utcnow().isoformat() + "Z"
|
||||
PROJECTS[pid] = {
|
||||
"id": pid,
|
||||
"contract": body.contract.model_dump() if body.contract else {},
|
||||
"costControl": body.costControl or {},
|
||||
"receivable": body.receivable or {},
|
||||
"payable": body.payable or {},
|
||||
"other": body.other or {},
|
||||
"createdAt": now,
|
||||
"updatedAt": now,
|
||||
}
|
||||
OPERATION_LOGS[pid] = []
|
||||
return {"id": pid, "message": "创建成功"}
|
||||
|
||||
|
||||
# ---------- 更新项目 ----------
|
||||
@app.post("/api/projects/update")
|
||||
def project_update(body: ProjectUpdateBody, authorization: Optional[str] = Header(None)):
|
||||
_get_role(authorization)
|
||||
pid = body.id
|
||||
if pid not in PROJECTS:
|
||||
return JSONResponse(status_code=404, content={"code": 404, "message": "项目不存在"})
|
||||
p = PROJECTS[pid]
|
||||
if body.contract is not None:
|
||||
p["contract"] = {**(p.get("contract") or {}), **body.contract}
|
||||
if body.costControl is not None:
|
||||
p["costControl"] = {**(p.get("costControl") or {}), **body.costControl}
|
||||
if body.receivable is not None:
|
||||
p["receivable"] = {**(p.get("receivable") or {}), **body.receivable}
|
||||
if body.payable is not None:
|
||||
p["payable"] = {**(p.get("payable") or {}), **body.payable}
|
||||
if body.other is not None:
|
||||
p["other"] = {**(p.get("other") or {}), **body.other}
|
||||
p["updatedAt"] = datetime.utcnow().isoformat() + "Z"
|
||||
return {"id": pid, "message": "保存成功"}
|
||||
|
||||
|
||||
# ---------- 操作日志 ----------
|
||||
@app.post("/api/projects/logs")
|
||||
def project_logs(body: ProjectLogsBody, authorization: Optional[str] = Header(None)):
|
||||
_get_role(authorization)
|
||||
pid = body.id
|
||||
if pid not in PROJECTS:
|
||||
return JSONResponse(status_code=404, content={"code": 404, "message": "项目不存在"})
|
||||
logs = OPERATION_LOGS.get(pid, [])
|
||||
page = body.page or 1
|
||||
page_size = body.pageSize or 20
|
||||
return {"list": logs, "total": len(logs), "page": page, "pageSize": page_size}
|
||||
Reference in New Issue
Block a user