ajyy/backend/tests/test_admin_feedback.py

"""管理端台账 / 办理 / 配置 API 测试"""
import pytest

from app.core.security import hash_password
from app.crud import user as user_crud
from app.main import app as fastapi_app
from app.core.database import get_db


def _login(client, username: str, password: str) -> dict:
    r = client.post(
        "/api/web/auth/login", json={"username": username, "password": password}
    )
    assert r.status_code == 200, r.text
    token = r.json()["data"]["token"]
    return {"Authorization": f"Bearer {token}"}


def _ensure_role_user(client, username: str, role_code: str, password: str = "Pass@123"):
    """在 client fixture 共享的 session 上插入角色用户。"""
    # 取出 override 的 session
    gen = fastapi_app.dependency_overrides[get_db]()
    db = next(gen)
    role = user_crud.get_role_by_code(db, role_code)
    assert role is not None
    existing = user_crud.get_user_by_username(db, username)
    if not existing:
        user_crud.create_user(
            db,
            username=username,
            password_hash=hash_password(password),
            display_name=username,
            role_id=role.id,
        )
    return _login(client, username, password)


@pytest.fixture()
def admin_headers(client):
    return _login(client, "admin", "Admin@123")


@pytest.fixture()
def handler_headers(client):
    return _ensure_role_user(client, "handler1", "handler")


@pytest.fixture()
def viewer_headers(client):
    return _ensure_role_user(client, "viewer1", "viewer")


def test_viewer_cannot_handle(client, viewer_headers):
    fb = client.post(
        "/api/public/feedback", json={"fbType": "投诉", "content": "xxxx"}
    ).json()["data"]
    r = client.put(
        f"/api/web/admin/feedbacks/{fb['id']}/status",
        headers=viewer_headers,
        json={"status": "处理中"},
    )
    assert r.status_code == 403
    assert r.json()["code"] == 403


def test_handler_can_handle_feedback(client, handler_headers):
    fb = client.post(
        "/api/public/feedback", json={"fbType": "投诉", "content": "需要办理的内容"}
    ).json()["data"]
    r = client.put(
        f"/api/web/admin/feedbacks/{fb['id']}/status",
        headers=handler_headers,
        json={"status": "处理中", "handleRemark": "已接单"},
    )
    assert r.status_code == 200
    assert r.json()["data"]["status"] == "处理中"
    assert r.json()["data"]["handleRemark"] == "已接单"


def test_invalid_feedback_transition(client, handler_headers):
    fb = client.post(
        "/api/public/feedback", json={"fbType": "表扬", "content": "好评内容测试"}
    ).json()["data"]
    # 待处理不能直接回到非法状态（例如空跳转到不存在状态）
    r = client.put(
        f"/api/web/admin/feedbacks/{fb['id']}/status",
        headers=handler_headers,
        json={"status": "已办结"},
    )
    assert r.status_code == 200  # 待处理→已办结 允许
    r2 = client.put(
        f"/api/web/admin/feedbacks/{fb['id']}/status",
        headers=handler_headers,
        json={"status": "待处理"},
    )
    assert r2.status_code == 400


def test_handler_cannot_update_hotline(client, handler_headers):
    r = client.put(
        "/api/web/admin/hotline",
        headers=handler_headers,
        json={"phone": "10086"},
    )
    assert r.status_code == 403


def test_admin_update_hotline(client, admin_headers):
    r = client.put(
        "/api/web/admin/hotline",
        headers=admin_headers,
        json={"phone": "0551-88888888", "workTime": "工作日 9-17"},
    )
    assert r.status_code == 200
    assert r.json()["data"]["phone"] == "0551-88888888"
    pub = client.get("/api/public/hotline")
    assert pub.json()["data"]["phone"] == "0551-88888888"


def test_dashboard(client, admin_headers):
    client.post(
        "/api/public/feedback", json={"fbType": "投诉", "content": "仪表盘反馈一条"}
    )
    r = client.get("/api/web/admin/dashboard", headers=admin_headers)
    assert r.status_code == 200
    data = r.json()["data"]
    assert data["feedbackTotal"] >= 1
    assert "recentFeedback" in data
    assert "complaintPending" in data


def test_list_feedbacks(client, admin_headers):
    client.post(
        "/api/public/feedback", json={"fbType": "投诉", "content": "列表筛选关键字甲"}
    )
    r = client.get(
        "/api/web/admin/feedbacks",
        headers=admin_headers,
        params={"keyword": "关键字甲", "page": 1, "pageSize": 10},
    )
    assert r.status_code == 200
    data = r.json()["data"]
    assert data["total"] >= 1
    assert data["list"][0]["content"].find("关键字甲") >= 0


def test_suggestion_handle(client, handler_headers):
    sg = client.post(
        "/api/public/suggestion", json={"content": "一条建言内容足够长"}
    ).json()["data"]
    r = client.put(
        f"/api/web/admin/suggestions/{sg['id']}/status",
        headers=handler_headers,
        json={"status": "已采纳"},
    )
    assert r.status_code == 200
    assert r.json()["data"]["status"] == "已采纳"