backend/tests/test_bulk_doc_ops.py

"""知識庫文件批次操作測試（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)