backend/tests/test_attachments.py

"""§12.19 附件（D97 部分解禁：上傳允許、無下載；inline preview only）。"""
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
from app.routers import attachments as attach_router

PWD = "TestPw123"


@pytest.fixture()
def client():
    return TestClient(app)


@pytest.fixture(autouse=True)
def isolated_attach_dir(tmp_path, monkeypatch):
    """每測試一個獨立 attachments 目錄。"""
    monkeypatch.setenv("KBV5_ATTACHMENTS_DIR", str(tmp_path / "att"))
    yield


def _mkuser(role, username, dept="業務部"):
    c = get_conn()
    c.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()),
    )
    uid = c.execute("SELECT id FROM users WHERE username=?", (username,)).fetchone()["id"]
    c.execute("INSERT OR IGNORE INTO users_departments (user_id, department, is_primary, assigned_at) "
              "VALUES (?,?,1,?)", (uid, dept, _utcnow()))
    return uid


def _mkdoc(author_id, dept, title):
    now = _utcnow()
    cur = get_conn().execute(
        "INSERT INTO documents (title, category, tags, summary, body, author_id, owner_department, "
        "visibility, allowed_departments, access_list, status, created_at, updated_at) "
        "VALUES (?,?,?,?,?,?,?,?,?,?,?,?,?)",
        (title, "知識/筆記", "[]", "s", "b", author_id, dept,
         "department", "[]", "[]", "active", now, now),
    )
    return cur.lastrowid


def _login(c, u):
    return c.post("/api/auth/login", json={"username": u, "password": PWD}).json()["access_token"]


def _auth(t):
    return {"Authorization": f"Bearer {t}"}


def _upload(client, doc_id, tok, *, filename="test.pdf", body=b"%PDF-1.4 test",
             mime="application/pdf"):
    return client.post(
        f"/api/documents/{doc_id}/attachments?filename={filename}",
        headers={**_auth(tok), "Content-Type": mime},
        content=body,
    )


# ============ upload 基本 ============

def test_admin_upload_success(client):
    _mkuser("admin", "adm")
    did = _mkdoc(1, "業務部", "doc")
    tok = _login(client, "adm")
    r = _upload(client, did, tok)
    assert r.status_code == 201
    j = r.json()
    assert j["filename"] == "test.pdf"
    assert j["mime_type"] == "application/pdf"
    assert j["size_bytes"] == len(b"%PDF-1.4 test")
    assert j["preview_url"] == f"/api/documents/{did}/attachments/{j['id']}"


def test_user_can_upload_own_doc(client):
    uid = _mkuser("user", "u1")
    did = _mkdoc(uid, "業務部", "mydoc")
    tok = _login(client, "u1")
    assert _upload(client, did, tok).status_code == 201


def test_user_cannot_upload_others_doc(client):
    _mkuser("user", "u1")
    _mkuser("user", "u2")
    did = _mkdoc(2, "業務部", "other")
    tok = _login(client, "u1")
    assert _upload(client, did, tok).status_code == 403


def test_manager_upload_own_dept_only(client):
    _mkuser("manager", "m1", "業務部")
    own = _mkdoc(1, "業務部", "own")
    other = _mkdoc(1, "研發部", "other")
    tok = _login(client, "m1")
    assert _upload(client, own, tok).status_code == 201
    assert _upload(client, other, tok).status_code == 403


# ============ 限制 ============

def test_reject_oversize(client):
    _mkuser("admin", "adm")
    did = _mkdoc(1, "業務部", "doc")
    tok = _login(client, "adm")
    big = b"x" * (attach_router.MAX_FILE_SIZE + 1)
    r = _upload(client, did, tok, filename="big.pdf", body=big)
    assert r.status_code == 413
    assert r.json()["detail"]["error"] == "file_too_large"


def test_reject_empty_file(client):
    _mkuser("admin", "adm")
    did = _mkdoc(1, "業務部", "doc")
    tok = _login(client, "adm")
    r = _upload(client, did, tok, body=b"")
    assert r.status_code == 400
    assert r.json()["detail"]["error"] == "empty_file"


def test_reject_unsupported_mime(client):
    _mkuser("admin", "adm")
    did = _mkdoc(1, "業務部", "doc")
    tok = _login(client, "adm")
    r = _upload(client, did, tok, filename="x.exe", body=b"MZ", mime="application/x-msdownload")
    assert r.status_code == 415


def test_quota_exceeded(client, monkeypatch):
    # monkeypatch 縮小上限（單篇總計已升 200MB，避免上傳數百 MB 巨檔）
    monkeypatch.setattr(attach_router, "MAX_FILE_SIZE", 2048)
    monkeypatch.setattr(attach_router, "MAX_TOTAL_PER_DOC", 4096)
    _mkuser("admin", "adm")
    did = _mkdoc(1, "業務部", "doc")
    tok = _login(client, "adm")
    one = b"x" * 2048    # 兩個 = 4096 = 上限
    assert _upload(client, did, tok, filename="a.pdf", body=one).status_code == 201
    assert _upload(client, did, tok, filename="b.pdf", body=one).status_code == 201
    r3 = _upload(client, did, tok, filename="c.pdf", body=b"x" * 16)
    assert r3.status_code == 413
    assert r3.json()["detail"]["error"] == "quota_exceeded"


def test_reject_pii_in_filename(client):
    # 測試環境 KBV5_SEED=0，需手動 seed PII 規則才有 tw_id 規則
    from app.db import seed_detection_rules
    seed_detection_rules(get_conn())
    _mkuser("admin", "adm")
    did = _mkdoc(1, "業務部", "doc")
    tok = _login(client, "adm")
    # 內建身分證 PII 規則命中
    r = _upload(client, did, tok, filename="A123456789_資料.pdf")
    assert r.status_code == 400
    assert r.json()["detail"]["error"] == "pii_detected"


# ============ list / preview / delete ============

def test_list_attachments(client):
    _mkuser("admin", "adm")
    did = _mkdoc(1, "業務部", "doc")
    tok = _login(client, "adm")
    _upload(client, did, tok, filename="a.pdf")
    _upload(client, did, tok, filename="b.pdf")
    r = client.get(f"/api/documents/{did}/attachments", headers=_auth(tok))
    assert r.status_code == 200
    j = r.json()
    assert j["count"] == 2
    assert j["quota_bytes"] == attach_router.MAX_TOTAL_PER_DOC


def test_preview_inline_no_download_header(client):
    """D97 鐵則：Content-Disposition 必為 inline，永不為 attachment。"""
    _mkuser("admin", "adm")
    did = _mkdoc(1, "業務部", "doc")
    tok = _login(client, "adm")
    up = _upload(client, did, tok, body=b"%PDF inline test").json()
    r = client.get(f"/api/documents/{did}/attachments/{up['id']}", headers=_auth(tok))
    assert r.status_code == 200
    cd = r.headers.get("content-disposition", "")
    assert cd == "inline"
    assert "attachment" not in cd
    assert r.content == b"%PDF inline test"
    assert r.headers["content-type"].startswith("application/pdf")


def test_preview_respects_can_view(client):
    """無權看 doc 的人 → 404（不洩漏存在）。"""
    _mkuser("admin", "adm")
    _mkuser("user", "u1", "客服部")
    did = _mkdoc(1, "業務部", "biz-only")
    adm_tok = _login(client, "adm")
    up = _upload(client, did, adm_tok).json()
    u1_tok = _login(client, "u1")
    r = client.get(f"/api/documents/{did}/attachments/{up['id']}", headers=_auth(u1_tok))
    assert r.status_code == 404


def test_uploader_can_delete(client):
    uid = _mkuser("user", "u1")
    did = _mkdoc(uid, "業務部", "mydoc")
    tok = _login(client, "u1")
    up = _upload(client, did, tok).json()
    r = client.delete(f"/api/documents/{did}/attachments/{up['id']}", headers=_auth(tok))
    assert r.status_code == 200
    assert r.json()["ok"] is True
    # 實體 + 元資料都清掉
    assert get_conn().execute("SELECT COUNT(*) AS c FROM attachments WHERE id=?",
                               (up["id"],)).fetchone()["c"] == 0


def test_other_user_cannot_delete(client):
    _mkuser("user", "u1")
    _mkuser("user", "u2")
    did = _mkdoc(1, "業務部", "doc")
    _mkuser("admin", "adm")
    adm_tok = _login(client, "adm")
    up = _upload(client, did, adm_tok).json()
    u2_tok = _login(client, "u2")
    r = client.delete(f"/api/documents/{did}/attachments/{up['id']}", headers=_auth(u2_tok))
    assert r.status_code == 403


# ============ 無 downloadAPI 端點存在性 ============

def test_no_download_endpoint_exists():
    """確認程式碼裡沒有任何 'download' 路徑（D97 鐵則：上傳允許、下載禁止）。"""
    paths = []
    for r in app.routes:
        if hasattr(r, "path") and "attachment" in r.path.lower():
            paths.append((r.path, sorted(getattr(r, "methods", []))))
    # 不存在 /download 子路徑
    assert all("download" not in p.lower() for p, _ in paths)
    # 只存在 GET / POST / DELETE
    all_methods = {m for _, ms in paths for m in ms}
    assert all_methods.issubset({"GET", "POST", "DELETE", "HEAD", "OPTIONS"})


# ============ ⑩#7 媒體庫分類計數 type_counts ============

def test_overview_returns_type_counts(client):
    """無 mime_prefix 時回 type_counts（圖片/PDF/影片/文件各分類可見數）。"""
    adm = _mkuser("admin", "adm")
    did = _mkdoc(adm, "業務部", "doc-tc")
    tok = _login(client, "adm")
    _upload(client, did, tok, filename="a.pdf", mime="application/pdf")
    _upload(client, did, tok, filename="b.png", body=b"\x89PNG test", mime="image/png")
    _upload(client, did, tok, filename="c.png", body=b"\x89PNG test2", mime="image/png")
    r = client.get("/api/attachments", headers=_auth(tok))
    assert r.status_code == 200, r.text
    tc = r.json().get("type_counts")
    assert tc is not None, "首載應回 type_counts"
    assert tc["pdf"] == 1 and tc["image"] == 2
    assert tc["all"] == tc["image"] + tc["pdf"] + tc["video"] + tc["doc"]


def test_overview_type_counts_absent_when_filtered(client):
    """帶 mime_prefix（過濾）時不回 type_counts（避免每次過濾重算）。"""
    _mkuser("admin", "adm")
    tok = _login(client, "adm")
    r = client.get("/api/attachments?mime_prefix=image/", headers=_auth(tok))
    assert r.status_code == 200
    assert "type_counts" not in r.json()


def test_type_counts_respects_visibility(client):
    """type_counts 須尊重可見性：看不到的文件其附件不計入。"""
    adm = _mkuser("admin", "adm")
    hidden = _mkdoc(adm, "研發部", "hidden")   # 研發部 department 文件
    atok = _login(client, "adm")
    _upload(client, hidden, atok, filename="x.png", body=b"\x89PNG", mime="image/png")
    _mkuser("user", "low", "業務部")           # 業務部 一般使用者：看不到研發部 department 文件
    ltok = _login(client, "low")
    r = client.get("/api/attachments", headers=_auth(ltok))
    tc = r.json()["type_counts"]
    assert tc["image"] == 0 and tc["all"] == 0, "看不到的附件不應計入 type_counts"


def test_type_counts_doc_category(client):
    """type_counts 的 doc（fallback）分類：docx 上傳後計入 doc；video 不可上傳故恆 0。"""
    adm = _mkuser("admin", "adm")
    did = _mkdoc(adm, "業務部", "d")
    tok = _login(client, "adm")
    _upload(client, did, tok, filename="d.docx", body=b"PKzip",
            mime="application/vnd.openxmlformats-officedocument.wordprocessingml.document")
    r = client.get("/api/attachments", headers=_auth(tok))
    tc = r.json()["type_counts"]
    assert tc["doc"] == 1 and tc["video"] == 0


def test_mime_cat_classification():
    """_mime_cat 純函式四分類（video 分支無法經上傳覆蓋 → 直接單測）。"""
    from app.routers.attachments import _mime_cat
    assert _mime_cat("image/png") == "image"
    assert _mime_cat("application/pdf") == "pdf"
    assert _mime_cat("video/mp4") == "video"
    assert _mime_cat("application/zip") == "doc"
    assert _mime_cat(None) == "doc"


# ============ 影片上傳（開放 mp4/webm/ogg，單檔 100MB）============

def test_video_mp4_upload_now_allowed(client):
    """mp4 影片現在可上傳（原本 415）。"""
    adm = _mkuser("admin", "adm")
    did = _mkdoc(adm, "業務部", "vd")
    tok = _login(client, "adm")
    r = _upload(client, did, tok, filename="clip.mp4", body=b"\x00\x00\x00\x18ftypmp42",
                mime="video/mp4")
    assert r.status_code == 201, r.text


def test_video_webm_ogg_allowed(client):
    """webm / ogg 影片可上傳。"""
    adm = _mkuser("admin", "adm")
    did = _mkdoc(adm, "業務部", "vd")
    tok = _login(client, "adm")
    assert _upload(client, did, tok, filename="a.webm", body=b"webmdata",
                   mime="video/webm").status_code == 201
    assert _upload(client, did, tok, filename="b.ogv", body=b"oggdata",
                   mime="video/ogg").status_code == 201


def test_video_avi_still_rejected(client):
    """非原生格式（avi）仍擋（只開放 mp4/webm/ogg）。"""
    adm = _mkuser("admin", "adm")
    did = _mkdoc(adm, "業務部", "vd")
    tok = _login(client, "adm")
    r = _upload(client, did, tok, filename="x.avi", body=b"avidata",
                mime="video/x-msvideo")
    assert r.status_code == 415


def test_video_constants_are_100mb_and_200mb_total():
    """確認上限常數：影片 100MB、單篇總計 200MB（使用者裁決）。"""
    assert attach_router.MAX_VIDEO_SIZE == 100 * 1024 * 1024
    assert attach_router.MAX_TOTAL_PER_DOC == 200 * 1024 * 1024


def test_video_over_limit_rejected(client, monkeypatch):
    """影片超過影片上限 → 413（monkeypatch 縮小上限避免上傳 100MB 巨檔）。"""
    monkeypatch.setattr(attach_router, "MAX_VIDEO_SIZE", 10)
    adm = _mkuser("admin", "adm")
    did = _mkdoc(adm, "業務部", "vd")
    tok = _login(client, "adm")
    r = _upload(client, did, tok, filename="big.mp4", body=b"\x00" * 20, mime="video/mp4")
    assert r.status_code == 413
    assert r.json()["detail"]["error"] == "file_too_large"


def test_video_uses_separate_cap_from_other_files(client, monkeypatch):
    """影片走影片上限、非影片走一般上限：同樣 20 bytes，非影片擋、影片過。"""
    monkeypatch.setattr(attach_router, "MAX_FILE_SIZE", 10)    # 一般檔上限縮小
    monkeypatch.setattr(attach_router, "MAX_VIDEO_SIZE", 1000)  # 影片上限較大
    adm = _mkuser("admin", "adm")
    did = _mkdoc(adm, "業務部", "vd")
    tok = _login(client, "adm")
    # 20 bytes 的 png → 超過一般 10 bytes → 413
    assert _upload(client, did, tok, filename="x.png", body=b"\x89PNG" + b"\x00" * 16,
                   mime="image/png").status_code == 413
    # 20 bytes 的 mp4 → 在影片 1000 bytes 內 → 201
    assert _upload(client, did, tok, filename="x.mp4", body=b"\x00" * 20,
                   mime="video/mp4").status_code == 201


def test_video_counts_in_type_counts(client):
    """影片上傳後計入 type_counts.video（媒體庫『影片』tab 不再恆 0）。"""
    adm = _mkuser("admin", "adm")
    did = _mkdoc(adm, "業務部", "vd")
    tok = _login(client, "adm")
    _upload(client, did, tok, filename="c.mp4", body=b"\x00ftypmp42", mime="video/mp4")
    r = client.get("/api/attachments", headers=_auth(tok))
    assert r.json()["type_counts"]["video"] == 1


# ============ 上傳/服務 安全回歸（釘住現在的安全行為，防未來重構破壞）============

def test_serve_sets_strict_csp_and_nosniff(client):
    """附件服務必含 nosniff + 嚴格 CSP(sandbox/script-src none) → SVG/HTML 即使被當文件也不執行 script。"""
    _mkuser("admin", "adm")
    did = _mkdoc(1, "業務部", "doc")
    tok = _login(client, "adm")
    up = _upload(client, did, tok).json()
    r = client.get(f"/api/documents/{did}/attachments/{up['id']}", headers=_auth(tok))
    assert r.status_code == 200
    assert r.headers.get("x-content-type-options") == "nosniff"
    csp = r.headers.get("content-security-policy", "")
    assert "script-src 'none'" in csp and "sandbox" in csp
    assert r.headers.get("content-disposition") == "inline"


def test_svg_with_script_served_neutralized(client):
    """惡意 SVG（內嵌 <script>）可上傳，但服務 header 中和（CSP 擋 script）。"""
    _mkuser("admin", "adm")
    did = _mkdoc(1, "業務部", "doc")
    tok = _login(client, "adm")
    evil = b'<svg xmlns="http://www.w3.org/2000/svg"><script>alert(document.cookie)</script></svg>'
    up = _upload(client, did, tok, filename="x.svg", body=evil, mime="image/svg+xml")
    assert up.status_code == 201
    r = client.get(f"/api/documents/{did}/attachments/{up.json()['id']}", headers=_auth(tok))
    # 內容原樣（未消毒）但 CSP sandbox + script-src none 使其無法執行
    assert "script-src 'none'" in r.headers.get("content-security-policy", "")
    assert r.headers.get("content-type", "").startswith("image/svg+xml")


def test_preview_requires_auth(client):
    """附件服務必須帶 token（無 token → 401）→ 無法直接以文件開啟 URL。"""
    _mkuser("admin", "adm")
    did = _mkdoc(1, "業務部", "doc")
    tok = _login(client, "adm")
    up = _upload(client, did, tok).json()
    r = client.get(f"/api/documents/{did}/attachments/{up['id']}")  # 無 Authorization
    assert r.status_code == 401


def test_html_disguised_as_png_not_executable(client):
    """content-type 偽造：HTML 內容宣稱 image/png → 服務以宣告型別 + nosniff + CSP，瀏覽器不會當 HTML 執行。"""
    _mkuser("admin", "adm")
    did = _mkdoc(1, "業務部", "doc")
    tok = _login(client, "adm")
    html = b"<html><script>alert(1)</script></html>"
    up = _upload(client, did, tok, filename="x.png", body=html, mime="image/png")
    assert up.status_code == 201
    r = client.get(f"/api/documents/{did}/attachments/{up.json()['id']}", headers=_auth(tok))
    assert r.headers.get("x-content-type-options") == "nosniff"
    assert r.headers.get("content-type", "").startswith("image/png")  # 不會被當 text/html
    assert "script-src 'none'" in r.headers.get("content-security-policy", "")