backend/tests/test_l8_admin.py

"""L8 管理面 B：敏感主題 / 機密樣本 / 白名單 / 變體詞典（PRD §49.11）。

需 embedding 的測試會在模型不可用時 skip。
"""
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.services import embeddings, pii_detector

PWD = "TestPw123"


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


@pytest.fixture()
def emb_on(monkeypatch):
    monkeypatch.setenv("KBV5_DISABLE_EMBEDDINGS", "0")
    embeddings.reset_for_tests()
    if not embeddings.is_available():
        pytest.skip("embedding 模型不可用")
    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()),
    )
    return c.execute("SELECT id FROM users WHERE username=?", (username,)).fetchone()["id"]


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_topic_create_and_block(client, emb_on):
    _mkuser("admin", "adm")
    tok = _login(client, "adm")
    r = client.post("/api/admin/sensitive-topics", headers=_auth(tok), json={
        "name": "未來技術藍圖", "threshold": 0.6,
        "samples": ["未來三年的研發技術藍圖與重點方向",
                    "下一代產品的技術路線規劃草案",
                    "尚未公開的長期技術發展策略"],
    })
    assert r.status_code == 201
    # 語意相近內容應被層 3 擋
    hit = pii_detector.scan(get_conn(), "這是我們未來三年的技術藍圖與研發路線重點規劃方向")
    assert hit and hit["layer"] == 3


def test_topic_need_3_samples(client, emb_on):
    _mkuser("admin", "adm")
    tok = _login(client, "adm")
    r = client.post("/api/admin/sensitive-topics", headers=_auth(tok),
                    json={"name": "x", "samples": ["只有一段"]})
    assert r.status_code == 400


def test_topic_builtin_not_deletable(client):
    _mkuser("admin", "adm")
    tok = _login(client, "adm")
    c = get_conn()
    c.execute("INSERT INTO sensitive_topics (name, is_builtin, is_active, threshold, created_at) "
              "VALUES ('financial',1,1,0.65,?)", (_utcnow(),))
    tid = c.execute("SELECT id FROM sensitive_topics WHERE name='financial'").fetchone()["id"]
    assert client.delete(f"/api/admin/sensitive-topics/{tid}", headers=_auth(tok)).status_code == 403


# ---------- 機密樣本 ----------

def test_conf_sample_add_block_dedup(client, emb_on):
    _mkuser("admin", "adm")
    tok = _login(client, "adm")
    txt = "本公司甘油保濕配方比例約八成並含關鍵原料供應商完整名單"
    r = client.post("/api/admin/confidential-samples", headers=_auth(tok),
                    json={"label": "配方v3", "text": txt, "step_up_password": PWD})
    assert r.status_code == 201
    # 列表不含原文 / embedding
    lst = client.get("/api/admin/confidential-samples", headers=_auth(tok)).json()
    assert lst and "embedding" not in lst[0] and "text" not in lst[0]
    # 相似內容被層 4 擋
    hit = pii_detector.scan(get_conn(), "我們甘油保濕配方大概八成左右還有主要原料供應商的名單")
    assert hit and hit["layer"] == 4
    # 重複上傳同文 → 409
    r2 = client.post("/api/admin/confidential-samples", headers=_auth(tok),
                     json={"label": "dup", "text": txt, "step_up_password": PWD})
    assert r2.status_code == 409


def test_conf_sample_wrong_stepup(client, emb_on):
    _mkuser("admin", "adm")
    tok = _login(client, "adm")
    r = client.post("/api/admin/confidential-samples", headers=_auth(tok),
                    json={"label": "x", "text": "機密內容測試文字段落", "step_up_password": "WRONGpw9"})
    assert r.status_code == 403


# ---------- 白名單 ----------

def test_whitelist_add_delete(client, emb_on):
    _mkuser("admin", "adm")
    tok = _login(client, "adm")
    r = client.post("/api/admin/whitelist-samples", headers=_auth(tok),
                    json={"label": "公開範本", "text": "這是一段公開可分享的標準範本內容文字"})
    assert r.status_code == 201
    wid = r.json()["id"]
    assert client.delete(f"/api/admin/whitelist-samples/{wid}", headers=_auth(tok)).status_code == 200


# ---------- 變體詞典 ----------

def test_variants_get_put_recompute(client):
    _mkuser("admin", "adm")
    tok = _login(client, "adm")
    # 先建一個 keyword 規則
    rid = client.post("/api/admin/detection-rules", headers=_auth(tok),
                      json={"rule_name": "kw", "rule_type": "keyword", "pattern": "甘油"}).json()["id"]
    d = client.get("/api/admin/variants", headers=_auth(tok)).json()
    d.setdefault("synonyms", {})["甘油"] = ["丙三醇", "保濕劑", "特殊新義詞XYZ"]
    r = client.put("/api/admin/variants", headers=_auth(tok), json={"data": d})
    assert r.status_code == 200 and r.json()["recomputed_rules"] >= 1
    # 規則變體應含新同義詞
    rule = next(x for x in client.get("/api/admin/detection-rules", headers=_auth(tok)).json() if x["id"] == rid)
    assert "特殊新義詞XYZ" in rule["variants"]


def test_variants_generate(client):
    _mkuser("admin", "adm")
    tok = _login(client, "adm")
    r = client.post("/api/admin/variants/generate", headers=_auth(tok), json={"term": "甘油"})
    assert r.status_code == 200
    j = r.json()
    # L10：mock 已換真實 pipeline；測試環境無金鑰 → 本地機械變體（source=local）
    assert j["source"] == "local" and isinstance(j["variants"], list)
    assert "甘 油" in j["variants"]  # 機械變體（拆字）