feat: yearly_targets 写权限收紧,12步冒烟全绿

2026-05-20 11:29:28 +08:00
parent 93fd327b13
commit 7cc0bf21ff
6 changed files with 521 additions and 8 deletions
@@ -22,9 +22,9 @@ router = APIRouter(prefix="/api/yearly_targets", tags=["年度目标"])
@router.get("", response_model=list[YearlyTargetResponse])
 def list_targets(
    session: Session = Depends(get_session),
-    current_user: User = Depends(require_role(UserRole.zhipianren, UserRole.zebian)),
+    current_user: User = Depends(require_role(UserRole.zhipianren, UserRole.zebian, UserRole.biandao)),
 ):
-    """获取所有年度目标（仅管理员/责编）。"""
+    """获取所有年度目标（三角色均可读）。"""
    statement = select(YearlyTarget).order_by(YearlyTarget.year.desc())
    targets = session.exec(statement).all()
    return targets
@@ -34,9 +34,9 @@ def list_targets(
 def get_target(
    target_id: int,
    session: Session = Depends(get_session),
-    current_user: User = Depends(require_role(UserRole.zhipianren, UserRole.zebian)),
+    current_user: User = Depends(require_role(UserRole.zhipianren, UserRole.zebian, UserRole.biandao)),
 ):
-    """根据 ID 获取单个年度目标（仅管理员/责编）。"""
+    """根据 ID 获取单个年度目标（三角色均可读）。"""
    target = session.get(YearlyTarget, target_id)
    if target is None:
        raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail="目标不存在")
@@ -101,9 +101,9 @@ def update_target(
 def delete_target(
    target_id: int,
    session: Session = Depends(get_session),
-    current_user: User = Depends(require_role(UserRole.zhipianren, UserRole.zebian)),
+    current_user: User = Depends(require_role(UserRole.zhipianren)),
 ):
-    """删除年度目标（仅管理员/责编）。"""
+    """删除年度目标（仅制片人）。"""
    target = session.get(YearlyTarget, target_id)
    if target is None:
        raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail="目标不存在")
@@ -21,8 +21,13 @@ class YearlyTargetCreate(YearlyTargetBase):
    pass


-class YearlyTargetUpdate(YearlyTargetBase):
-    pass
+class YearlyTargetUpdate(BaseModel):
+    """PATCH 时允许只传部分字段，year 不可修改（由 API 层 pop 掉）。"""
+    year: int | None = None
+    base_target: float | None = None
+    stretch_target: float | None = None
+    issued_date: date | None = None
+    notes: str | None = None


 class YearlyTargetResponse(YearlyTargetBase):
@@ -0,0 +1,419 @@
+"""
+yearly_targets 写权限收紧 — 冒烟测试脚本
+
+幂等，可重复跑。
+确保后端已启动: uvicorn app.main:app --reload
+
+12 步覆盖:
+1. zhipianren POST → 201
+2. zebian POST → 201
+3. biandao POST → 403
+4. 未登录 POST → 401
+5. zhipianren PATCH → 200
+6. zebian PATCH → 200
+7. biandao PATCH → 403
+8. zhipianren DELETE → 200
+9. zebian DELETE → 403
+10. biandao DELETE → 403
+11. biandao GET 列表 → 200
+12. zhipianren GET 列表 → 200
+"""
+
+import sys
+from pathlib import Path
+
+backend_dir = Path(__file__).resolve().parent.parent
+sys.path.insert(0, str(backend_dir))
+
+import httpx
+
+BASE = "http://localhost:8000"
+HEADERS = {"Content-Type": "application/json"}
+
+
+def login(username: str, password: str) -> str:
+    """登录并返回 session cookie。"""
+    r = httpx.post(f"{BASE}/api/auth/login", json={"username": username, "password": password})
+    r.raise_for_status()
+    return r.cookies.get("milsci_session", "")
+
+
+def step(name: str, expected_status: int, method: str, url: str, **kwargs):
+    """执行一个 HTTP 步骤，比较实际状态码与期望值。"""
+    cookies = kwargs.pop("cookies", {})
+    if method == "GET":
+        r = httpx.get(url, cookies=cookies, timeout=10)
+    elif method == "POST":
+        r = httpx.post(url, cookies=cookies, headers=HEADERS, timeout=10, **kwargs)
+    elif method == "PATCH":
+        r = httpx.patch(url, cookies=cookies, headers=HEADERS, timeout=10, **kwargs)
+    elif method == "DELETE":
+        r = httpx.delete(url, cookies=cookies, timeout=10)
+    else:
+        raise ValueError(method)
+
+    status_ok = r.status_code == expected_status
+    mark = "[OK]" if status_ok else f"[FAIL expected {expected_status} got {r.status_code}]"
+    print(f"  {mark} [{method}] {name}")
+    if not status_ok:
+        print(f"       Response: {r.text[:300]}")
+    return r, status_ok
+
+
+def ensure_user(cookies: dict, username: str, display_name: str, role: str, password: str = "Test1234"):
+    """幂等创建测试账号（如果不存在）。"""
+    # 先尝试登录（账号可能已存在）
+    try:
+        r = httpx.get(f"{BASE}/api/users/me", cookies=cookies, timeout=5)
+        if r.status_code == 200:
+            me = r.json()
+            if me.get("username") == username:
+                print(f"  [SKIP] user {username} already exists, reusing")
+                return
+    except Exception:
+        pass
+
+    # 尝试创建
+    r = httpx.post(
+        f"{BASE}/api/users",
+        json={"username": username, "display_name": display_name, "password": password, "role": role},
+        cookies=cookies,
+        headers=HEADERS,
+        timeout=10,
+    )
+    if r.status_code == 201:
+        print(f"  [OK] created {username} ({role})")
+    elif r.status_code == 409:
+        print(f"  [SKIP] {username} already exists")
+    else:
+        print(f"  [WARN] create {username} returned {r.status_code}: {r.text[:100]}")
+
+
+def cleanup_test_target(cookies: dict):
+    """清理测试用年度目标（2027），幂等。"""
+    try:
+        r = httpx.get(f"{BASE}/api/yearly_targets", cookies=cookies, timeout=10)
+        if r.status_code == 200:
+            for t in r.json():
+                if t.get("year") == 2027:
+                    httpx.delete(f"{BASE}/api/yearly_targets/{t['id']}", cookies=cookies, timeout=10)
+                    print(f"  [CLEAN] removed existing 2027 target ID={t['id']}")
+    except Exception:
+        pass
+
+
+def main():
+    print("=" * 60)
+    print("  yearly_targets 写权限收紧 — 冒烟测试")
+    print("=" * 60)
+
+    # Step 0: 用 zhipianren 登录，准备测试账号
+    print("\n[Step 0] Login as zhipianren and ensure test accounts exist")
+    try:
+        admin_cookie = login("simonkoson", "liutong65")
+    except Exception as e:
+        print(f"  [FAIL] cannot login as simonkoson: {e}")
+        return
+
+    admin_cookies = {"milsci_session": admin_cookie}
+    print("  [OK] logged in as simonkoson (zhipianren)")
+
+    # 幂等创建 zebian / biandao 测试账号
+    ensure_user(admin_cookies, "test_zebian_smoke", "TestZebra Smoke", "zebian")
+    ensure_user(admin_cookies, "test_biandao_smoke", "TestBiandao Smoke", "biandao")
+
+    # 登录 zebian 和 biandao 获取 cookie
+    print("\n[Step 0b] Login as zebian and biandao")
+    zebian_cookie = login("test_zebian_smoke", "Test1234")
+    zebian_cookies = {"milsci_session": zebian_cookie}
+    print("  [OK] logged in as test_zebian_smoke (zebian)")
+
+    biandao_cookie = login("test_biandao_smoke", "Test1234")
+    biandao_cookies = {"milsci_session": biandao_cookie}
+    print("  [OK] logged in as test_biandao_smoke (biandao)")
+
+    # 清理可能残留的 2027 目标
+    cleanup_test_target(admin_cookies)
+
+    results = []
+
+    # =====================================================================
+    # Step 1: zhipianren POST → 201
+    # =====================================================================
+    print("\n[Step 1] POST new yearly target — zhipianren → 201")
+    r, ok = step(
+        "POST /api/yearly_targets (zhipianren)",
+        201,
+        "POST",
+        f"{BASE}/api/yearly_targets",
+        json={"year": 2027, "base_target": 0.66, "stretch_target": 0.91},
+        cookies=admin_cookies,
+    )
+    target_id = r.json().get("id") if ok else None
+    results.append(("Step 1: zhipianren POST 201", ok))
+
+    # =====================================================================
+    # Step 2: zebian POST → 201
+    # =====================================================================
+    print("\n[Step 2] POST new yearly target — zebian → 201")
+    # 清理 2027（刚被 zhipianren 创建了）
+    cleanup_test_target(admin_cookies)
+    r, ok = step(
+        "POST /api/yearly_targets (zebian)",
+        201,
+        "POST",
+        f"{BASE}/api/yearly_targets",
+        json={"year": 2027, "base_target": 0.66, "stretch_target": 0.91},
+        cookies=zebian_cookies,
+    )
+    target_id2 = r.json().get("id") if ok else None
+    results.append(("Step 2: zebian POST 201", ok))
+
+    # =====================================================================
+    # Step 3: biandao POST → 403
+    # =====================================================================
+    print("\n[Step 3] POST new yearly target — biandao → 403")
+    cleanup_test_target(admin_cookies)
+    r, ok = step(
+        "POST /api/yearly_targets (biandao)",
+        403,
+        "POST",
+        f"{BASE}/api/yearly_targets",
+        json={"year": 2027, "base_target": 0.66, "stretch_target": 0.91},
+        cookies=biandao_cookies,
+    )
+    results.append(("Step 3: biandao POST 403", ok))
+
+    # =====================================================================
+    # Step 4: 未登录 POST → 401
+    # =====================================================================
+    print("\n[Step 4] POST new yearly target — not logged in → 401")
+    r, ok = step(
+        "POST /api/yearly_targets (no auth)",
+        401,
+        "POST",
+        f"{BASE}/api/yearly_targets",
+        json={"year": 2027, "base_target": 0.66, "stretch_target": 0.91},
+        cookies={},
+    )
+    results.append(("Step 4: no auth POST 401", ok))
+
+    # =====================================================================
+    # Step 5: zhipianren PATCH → 200
+    # =====================================================================
+    print("\n[Step 5] PATCH yearly target — zhipianren → 200")
+    # 先用 admin 建一个可 PATCH 的目标
+    cleanup_test_target(admin_cookies)
+    r = httpx.post(
+        f"{BASE}/api/yearly_targets",
+        json={"year": 2027, "base_target": 0.66, "stretch_target": 0.91},
+        cookies=admin_cookies,
+        headers=HEADERS,
+        timeout=10,
+    )
+    patch_id = r.json().get("id") if r.status_code == 201 else None
+
+    if patch_id:
+        r, ok = step(
+            "PATCH /api/yearly_targets/{id} (zhipianren)",
+            200,
+            "PATCH",
+            f"{BASE}/api/yearly_targets/{patch_id}",
+            json={"base_target": 0.67},
+            cookies=admin_cookies,
+        )
+    else:
+        ok = False
+        print("  [FAIL] could not create target for PATCH test")
+    results.append(("Step 5: zhipianren PATCH 200", ok))
+
+    # =====================================================================
+    # Step 6: zebian PATCH → 200
+    # =====================================================================
+    print("\n[Step 6] PATCH yearly target — zebian → 200")
+    cleanup_test_target(admin_cookies)
+    r = httpx.post(
+        f"{BASE}/api/yearly_targets",
+        json={"year": 2027, "base_target": 0.66, "stretch_target": 0.91},
+        cookies=admin_cookies,
+        headers=HEADERS,
+        timeout=10,
+    )
+    patch_id2 = r.json().get("id") if r.status_code == 201 else None
+
+    if patch_id2:
+        r, ok = step(
+            "PATCH /api/yearly_targets/{id} (zebian)",
+            200,
+            "PATCH",
+            f"{BASE}/api/yearly_targets/{patch_id2}",
+            json={"stretch_target": 0.92},
+            cookies=zebian_cookies,
+        )
+    else:
+        ok = False
+        print("  [FAIL] could not create target for PATCH test")
+    results.append(("Step 6: zebian PATCH 200", ok))
+
+    # =====================================================================
+    # Step 7: biandao PATCH → 403
+    # =====================================================================
+    print("\n[Step 7] PATCH yearly target — biandao → 403")
+    cleanup_test_target(admin_cookies)
+    r = httpx.post(
+        f"{BASE}/api/yearly_targets",
+        json={"year": 2027, "base_target": 0.66, "stretch_target": 0.91},
+        cookies=admin_cookies,
+        headers=HEADERS,
+        timeout=10,
+    )
+    patch_id3 = r.json().get("id") if r.status_code == 201 else None
+
+    if patch_id3:
+        r, ok = step(
+            "PATCH /api/yearly_targets/{id} (biandao)",
+            403,
+            "PATCH",
+            f"{BASE}/api/yearly_targets/{patch_id3}",
+            json={"base_target": 0.68},
+            cookies=biandao_cookies,
+        )
+    else:
+        ok = False
+        print("  [FAIL] could not create target for PATCH test")
+    results.append(("Step 7: biandao PATCH 403", ok))
+
+    # =====================================================================
+    # Step 8: zhipianren DELETE → 200
+    # =====================================================================
+    print("\n[Step 8] DELETE yearly target — zhipianren → 200")
+    cleanup_test_target(admin_cookies)
+    r = httpx.post(
+        f"{BASE}/api/yearly_targets",
+        json={"year": 2027, "base_target": 0.66, "stretch_target": 0.91},
+        cookies=admin_cookies,
+        headers=HEADERS,
+        timeout=10,
+    )
+    del_id = r.json().get("id") if r.status_code == 201 else None
+
+    if del_id:
+        r, ok = step(
+            "DELETE /api/yearly_targets/{id} (zhipianren)",
+            204,
+            "DELETE",
+            f"{BASE}/api/yearly_targets/{del_id}",
+            cookies=admin_cookies,
+        )
+    else:
+        ok = False
+        print("  [FAIL] could not create target for DELETE test")
+    results.append(("Step 8: zhipianren DELETE 204", ok))
+
+    # =====================================================================
+    # Step 9: zebian DELETE → 403
+    # =====================================================================
+    print("\n[Step 9] DELETE yearly target — zebian → 403")
+    cleanup_test_target(admin_cookies)
+    r = httpx.post(
+        f"{BASE}/api/yearly_targets",
+        json={"year": 2027, "base_target": 0.66, "stretch_target": 0.91},
+        cookies=admin_cookies,
+        headers=HEADERS,
+        timeout=10,
+    )
+    del_id2 = r.json().get("id") if r.status_code == 201 else None
+
+    if del_id2:
+        r, ok = step(
+            "DELETE /api/yearly_targets/{id} (zebian)",
+            403,
+            "DELETE",
+            f"{BASE}/api/yearly_targets/{del_id2}",
+            cookies=zebian_cookies,
+        )
+    else:
+        ok = False
+        print("  [FAIL] could not create target for DELETE test")
+    results.append(("Step 9: zebian DELETE 403", ok))
+
+    # =====================================================================
+    # Step 10: biandao DELETE → 403
+    # =====================================================================
+    print("\n[Step 10] DELETE yearly target — biandao → 403")
+    cleanup_test_target(admin_cookies)
+    r = httpx.post(
+        f"{BASE}/api/yearly_targets",
+        json={"year": 2027, "base_target": 0.66, "stretch_target": 0.91},
+        cookies=admin_cookies,
+        headers=HEADERS,
+        timeout=10,
+    )
+    del_id3 = r.json().get("id") if r.status_code == 201 else None
+
+    if del_id3:
+        r, ok = step(
+            "DELETE /api/yearly_targets/{id} (biandao)",
+            403,
+            "DELETE",
+            f"{BASE}/api/yearly_targets/{del_id3}",
+            cookies=biandao_cookies,
+        )
+    else:
+        ok = False
+        print("  [FAIL] could not create target for DELETE test")
+    results.append(("Step 10: biandao DELETE 403", ok))
+
+    # =====================================================================
+    # Step 11: biandao GET 列表 → 200
+    # =====================================================================
+    print("\n[Step 11] GET yearly targets list — biandao → 200")
+    r, ok = step(
+        "GET /api/yearly_targets (biandao)",
+        200,
+        "GET",
+        f"{BASE}/api/yearly_targets",
+        cookies=biandao_cookies,
+    )
+    results.append(("Step 11: biandao GET 200", ok))
+
+    # =====================================================================
+    # Step 12: zhipianren GET 列表 → 200
+    # =====================================================================
+    print("\n[Step 12] GET yearly targets list — zhipianren → 200")
+    r, ok = step(
+        "GET /api/yearly_targets (zhipianren)",
+        200,
+        "GET",
+        f"{BASE}/api/yearly_targets",
+        cookies=admin_cookies,
+    )
+    results.append(("Step 12: zhipianren GET 200", ok))
+
+    # =====================================================================
+    # Summary
+    # =====================================================================
+    print("\n" + "=" * 60)
+    print("  结果汇总")
+    print("=" * 60)
+    passed = sum(1 for _, ok in results if ok)
+    total = len(results)
+    for name, ok in results:
+        mark = "[OK]" if ok else "[FAIL]"
+        print(f"  {mark} {name}")
+
+    print(f"\n  PASS: {passed}/{total}")
+    if passed == total:
+        print("  ALL PASS")
+    else:
+        print("  SOME FAILURES")
+
+    print("\n测试账号已保留(供后续手动清理):")
+    print("  - test_zebian_smoke (zebian)")
+    print("  - test_biandao_smoke (biandao)")
+
+    print("\n" + "=" * 60)
+
+
+if __name__ == "__main__":
+    main()