"""知識庫文件批次操作測試（PRD §10.4 + 745 + D52 §10.4.1）。 涵蓋： - 批次改可見度（改為部門內部 / 改為公開）— 公開永遠 step-up（使用者裁決依 §10.4）； 部門內部僅在「擴大受眾」時 step-up（沿用單篇方向性 A6 §11.3）。 - 批次軟刪除 — 一律 step-up（§10.4 軟刪除文件須 re-confirm 密碼）。 - 安全：權限不足的文件**跳過**（fail closed，不誤改他人/他部門）；密碼錯 → 403。 安全模組 test-first（十二誡 §11）：先寫此檔，再實作 bulk 端點。 """ from __future__ import annotations import pytest from fastapi.testclient import TestClient from app.auth.passwords import hash_password from app.db import _utcnow, get_conn from app.main import app PWD = "TestPw123!" @pytest.fixture() def client(): return TestClient(app) def _mkuser(username, role, dept): get_conn().execute( "INSERT INTO users (username, email, password_hash, role, department, " "created_at, is_active) VALUES (?,?,?,?,?,?,1)", (username, f"{username}@x.tw", hash_password(PWD), role, dept, _utcnow()), ) get_conn().execute( "INSERT OR IGNORE INTO departments (name, sort_order, is_active, created_at) " "VALUES (?, 0, 1, ?)", (dept, _utcnow()), ) return get_conn().execute( "SELECT id FROM users WHERE username=?", (username,)).fetchone()["id"] _DOC_SEQ = [0] def _mkdoc(author_id, dept="業務部", vis="department"): now = _utcnow() _DOC_SEQ[0] += 1 cur = get_conn().execute( "INSERT INTO documents (title, category, tags, summary, body, author_id, " "owner_department, visibility, allowed_departments, access_list, status, " "vault_path, created_at, updated_at) " "VALUES (?,?,?,?,?,?,?,?,'[]','[]','active',?,?,?)", ("t", "知識/筆記", "[]", "s", "b", author_id, dept, vis, f"vault/bulk-{_DOC_SEQ[0]}.md", now, now), ) return cur.lastrowid def _login(c, username="adm"): return c.post("/api/auth/login", json={"username": username, "password": PWD}).json()["access_token"] def _auth(t): return {"Authorization": f"Bearer {t}"} def _vis(did): return get_conn().execute( "SELECT visibility FROM documents WHERE id=?", (did,)).fetchone()["visibility"] def _deleted(did): return get_conn().execute( "SELECT is_deleted FROM documents WHERE id=?", (did,)).fetchone()["is_deleted"] # ============ 批次改可見度 ============ def test_bulk_to_public_requires_step_up(client): """批次改公開 → 一律 step-up（使用者裁決依 §10.4）。""" uid = _mkuser("adm", "admin", "業務部") ids = [_mkdoc(uid), _mkdoc(uid), _mkdoc(uid)] tok = _login(client) # 無密碼 → 403 step_up_required r = client.post("/api/documents/bulk/visibility", json={"ids": ids, "visibility": "public"}, headers=_auth(tok)) assert r.status_code == 403, r.text assert r.json()["detail"]["error"] == "step_up_required" # 確認沒有任何文件被改（fail closed） assert all(_vis(d) == "department" for d in ids) # 帶密碼 → 200，全部改公開 r2 = client.post("/api/documents/bulk/visibility", json={"ids": ids, "visibility": "public", "step_up_password": PWD}, headers=_auth(tok)) assert r2.status_code == 200, r2.text assert r2.json()["count"] == 3 assert all(_vis(d) == "public" for d in ids) def test_bulk_to_public_wrong_password_rejected(client): """密碼錯 → 403 step_up_failed，且不改任何文件。""" uid = _mkuser("adm", "admin", "業務部") ids = [_mkdoc(uid), _mkdoc(uid)] tok = _login(client) r = client.post("/api/documents/bulk/visibility", json={"ids": ids, "visibility": "public", "step_up_password": "WRONG"}, headers=_auth(tok)) assert r.status_code == 403 assert r.json()["detail"]["error"] == "step_up_failed" assert all(_vis(d) == "department" for d in ids) def test_bulk_to_department_narrowing_no_password(client): """批次改部門內部：public → department 屬縮小受眾，免 step-up。""" uid = _mkuser("adm", "admin", "業務部") ids = [_mkdoc(uid, vis="public"), _mkdoc(uid, vis="public")] tok = _login(client) r = client.post("/api/documents/bulk/visibility", json={"ids": ids, "visibility": "department"}, headers=_auth(tok)) assert r.status_code == 200, r.text assert all(_vis(d) == "department" for d in ids) def test_bulk_to_department_expanding_requires_step_up(client): """批次改部門內部：confidential → department 屬擴大受眾，需 step-up。""" uid = _mkuser("adm", "admin", "業務部") ids = [_mkdoc(uid, vis="confidential")] tok = _login(client) r = client.post("/api/documents/bulk/visibility", json={"ids": ids, "visibility": "department"}, headers=_auth(tok)) assert r.status_code == 403 assert r.json()["detail"]["error"] == "step_up_required" def test_bulk_visibility_skips_unpermitted_docs(client): """manager 只能改本部門：他部門文件被跳過（fail closed），不誤改。""" adm = _mkuser("adm", "admin", "業務部") _mkuser("mgr", "manager", "業務部") mine = _mkdoc(adm, dept="業務部", vis="public") other = _mkdoc(adm, dept="研發部", vis="public") tok = _login(client, "mgr") r = client.post("/api/documents/bulk/visibility", json={"ids": [mine, other], "visibility": "department"}, headers=_auth(tok)) assert r.status_code == 200, r.text assert _vis(mine) == "department" # 本部門：改了 assert _vis(other) == "public" # 他部門：跳過 assert r.json()["count"] == 1 assert any(s["id"] == other for s in r.json()["skipped"]) # ============ 批次軟刪除 ============ def test_bulk_delete_requires_step_up(client): """批次軟刪除須 step-up（§10.4）。缺密碼 → 422（欄位必填）。""" uid = _mkuser("adm", "admin", "業務部") ids = [_mkdoc(uid), _mkdoc(uid)] tok = _login(client) r = client.post("/api/documents/bulk/delete", json={"ids": ids}, headers=_auth(tok)) assert r.status_code == 422 # step_up_password 必填 assert all(_deleted(d) == 0 for d in ids) def test_bulk_delete_wrong_password_rejected(client): """密碼錯 → 403，不刪任何文件。""" uid = _mkuser("adm", "admin", "業務部") ids = [_mkdoc(uid), _mkdoc(uid)] tok = _login(client) r = client.post("/api/documents/bulk/delete", json={"ids": ids, "step_up_password": "WRONG"}, headers=_auth(tok)) assert r.status_code == 403 assert all(_deleted(d) == 0 for d in ids) def test_bulk_delete_success(client): """正確密碼 → 全部軟刪除。""" uid = _mkuser("adm", "admin", "業務部") ids = [_mkdoc(uid), _mkdoc(uid), _mkdoc(uid)] tok = _login(client) r = client.post("/api/documents/bulk/delete", json={"ids": ids, "step_up_password": PWD}, headers=_auth(tok)) assert r.status_code == 200, r.text assert r.json()["count"] == 3 assert all(_deleted(d) == 1 for d in ids) def test_bulk_delete_skips_unpermitted(client): """manager 只能刪本部門：他部門被跳過。""" adm = _mkuser("adm", "admin", "業務部") _mkuser("mgr", "manager", "業務部") mine = _mkdoc(adm, dept="業務部") other = _mkdoc(adm, dept="研發部") tok = _login(client, "mgr") r = client.post("/api/documents/bulk/delete", json={"ids": [mine, other], "step_up_password": PWD}, headers=_auth(tok)) assert r.status_code == 200, r.text assert _deleted(mine) == 1 assert _deleted(other) == 0 assert r.json()["count"] == 1 # ============ 輸入防呆（#不信任輸入）============ def test_bulk_empty_ids_rejected(client): """空 ids → 400。""" _mkuser("adm", "admin", "業務部") tok = _login(client) r = client.post("/api/documents/bulk/visibility", json={"ids": [], "visibility": "department"}, headers=_auth(tok)) assert r.status_code == 400 def test_bulk_too_many_ids_rejected(client): """超過上限（200）→ 400，避免一次操作癱瘓。""" _mkuser("adm", "admin", "業務部") tok = _login(client) r = client.post("/api/documents/bulk/delete", json={"ids": list(range(1, 500)), "step_up_password": PWD}, headers=_auth(tok)) assert r.status_code == 400 def test_bulk_to_department_expanding_success_with_password(client): """批次改部門內部·擴大受眾路徑：帶正確密碼 → 200 並真的改成 department。""" uid = _mkuser("adm", "admin", "業務部") ids = [_mkdoc(uid, vis="confidential"), _mkdoc(uid, vis="restricted")] tok = _login(client) r = client.post("/api/documents/bulk/visibility", json={"ids": ids, "visibility": "department", "step_up_password": PWD}, headers=_auth(tok)) assert r.status_code == 200, r.text assert r.json()["count"] == 2 assert all(_vis(d) == "department" for d in ids)