backend/tests/test_helpful.py

"""§50.3.3 有用按鈕 endpoint 測試。"""
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(name, role="user", dept="業務部"):
    c = get_conn()
    c.execute(
        "INSERT INTO users (username, email, password_hash, role, department, "
        "created_at, is_active) VALUES (?,?,?,?,?,?,1)",
        (name, f"{name}@x.tw", hash_password(PWD), role, dept, _utcnow()),
    )
    return c.execute("SELECT id FROM users WHERE username=?", (name,)).fetchone()["id"]


def _mkdoc(author_id, vis="public", dept="業務部"):
    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 (?,?,?,?,?,?,?,?,'[]','[]','active',?,?)",
        ("t", "知識/筆記", "[]", "s", "b", author_id, dept, vis, 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 test_toggle_helpful_on(client):
    a = _mkuser("author")
    did = _mkdoc(a)
    u = _mkuser("voter")
    tok = _login(client, "voter")
    r = client.post(f"/api/documents/{did}/helpful", headers=_auth(tok))
    assert r.status_code == 200
    j = r.json()
    assert j["helpful"] is True
    assert j["count"] == 1


def test_toggle_helpful_off(client):
    a = _mkuser("author")
    did = _mkdoc(a)
    _mkuser("voter")
    tok = _login(client, "voter")
    client.post(f"/api/documents/{did}/helpful", headers=_auth(tok))
    r = client.post(f"/api/documents/{did}/helpful", headers=_auth(tok))  # 再點取消
    j = r.json()
    assert j["helpful"] is False
    assert j["count"] == 0


def test_helpful_count_multi_user(client):
    a = _mkuser("author")
    did = _mkdoc(a)
    for name in ("v1", "v2", "v3"):
        _mkuser(name)
        tok = _login(client, name)
        client.post(f"/api/documents/{did}/helpful", headers=_auth(tok))
    cnt = get_conn().execute("SELECT helpful_count FROM documents WHERE id=?", (did,)).fetchone()["helpful_count"]
    assert cnt == 3


def test_get_helpful_status(client):
    a = _mkuser("author")
    did = _mkdoc(a)
    _mkuser("voter")
    tok = _login(client, "voter")
    # 未投
    r1 = client.get(f"/api/documents/{did}/helpful", headers=_auth(tok))
    assert r1.json()["helpful"] is False
    # 投了
    client.post(f"/api/documents/{did}/helpful", headers=_auth(tok))
    r2 = client.get(f"/api/documents/{did}/helpful", headers=_auth(tok))
    assert r2.json()["helpful"] is True
    assert r2.json()["count"] == 1


def test_helpful_404_missing_doc(client):
    _mkuser("voter")
    tok = _login(client, "voter")
    r = client.post("/api/documents/99999/helpful", headers=_auth(tok))
    assert r.status_code == 404


def test_helpful_respects_visibility(client):
    a = _mkuser("author", dept="業務部")
    did = _mkdoc(a, vis="department", dept="業務部")
    _mkuser("outsider", dept="研發部")
    tok = _login(client, "outsider")
    r = client.post(f"/api/documents/{did}/helpful", headers=_auth(tok))
    assert r.status_code == 404  # 跨部門看不到 → 不能投


def test_helpful_unauth(client):
    a = _mkuser("author")
    did = _mkdoc(a)
    r = client.post(f"/api/documents/{did}/helpful")
    assert r.status_code in (401, 403)


# ============ helpful_count 不 drift（巡檢 #helpful）============

def test_helpful_count_always_equals_vote_rows(client):
    """stored helpful_count 永遠 == COUNT(helpful_votes)，且 toggle 冪等不漂移。"""
    aid = _mkuser("a1", "admin")
    _mkuser("u1", "user")
    _mkuser("u2", "user")
    did = _mkdoc(aid, "public", "業務部")

    def _stored_vs_real():
        c = get_conn()
        stored = c.execute("SELECT helpful_count FROM documents WHERE id=?", (did,)).fetchone()["helpful_count"]
        real = c.execute("SELECT COUNT(*) AS n FROM helpful_votes WHERE document_id=?", (did,)).fetchone()["n"]
        return stored, real

    t1 = _login(client, "u1"); t2 = _login(client, "u2")
    client.post(f"/api/documents/{did}/helpful", headers=_auth(t1))   # u1 讚
    client.post(f"/api/documents/{did}/helpful", headers=_auth(t2))   # u2 讚
    s, r = _stored_vs_real(); assert s == r == 2
    # u1 重複按（toggle off）
    client.post(f"/api/documents/{did}/helpful", headers=_auth(t1))
    s, r = _stored_vs_real(); assert s == r == 1
    # u1 再按回（toggle on）
    last = client.post(f"/api/documents/{did}/helpful", headers=_auth(t1)).json()
    s, r = _stored_vs_real(); assert s == r == 2
    assert last["count"] == 2 and last["helpful"] is True