81 lines
2.6 KiB
Python
81 lines
2.6 KiB
Python
|
# -*- coding: utf-8 -*-
|
||
|
|
"""运营端业务端点测试：任务/服务商/内容/配置/统计。"""
|
||
|
|
from __future__ import annotations
|
||
|
|
|
||
|
|
from fastapi.testclient import TestClient
|
||
|
|
|
||
|
|
|
||
|
|
def login(client: TestClient, username: str, password: str = "123456") -> str:
|
||
|
|
res = client.post("/auth/login", json={"username": username, "password": password})
|
||
|
|
assert res.status_code == 200, res.text
|
||
|
|
return res.json()["token"]
|
||
|
|
|
||
|
|
|
||
|
|
def auth(token: str) -> dict:
|
||
|
|
return {"Authorization": f"Bearer {token}"}
|
||
|
|
|
||
|
|
|
||
|
|
def test_operator_lists_seeded_tasks(client):
|
||
|
|
op = login(client, "pine")
|
||
|
|
tasks = client.get("/admin/tasks", headers=auth(op)).json()
|
||
|
|
assert len(tasks) >= 3
|
||
|
|
assert any(t["id"] == "task_001" for t in tasks)
|
||
|
|
|
||
|
|
|
||
|
|
def test_operator_task_status_flow(client):
|
||
|
|
op = login(client, "pine")
|
||
|
|
# 审核通过 task_003 (pending -> published)
|
||
|
|
res = client.post(
|
||
|
|
"/admin/tasks/task_003/status", headers=auth(op), json={"status": "published"},
|
||
|
|
)
|
||
|
|
assert res.status_code == 200, res.text
|
||
|
|
assert res.json()["status"] == "published"
|
||
|
|
|
||
|
|
|
||
|
|
def test_operator_provider_list_and_update(client):
|
||
|
|
op = login(client, "pine")
|
||
|
|
provs = client.get("/admin/providers", headers=auth(op)).json()
|
||
|
|
assert len(provs) >= 3
|
||
|
|
res = client.post(
|
||
|
|
"/admin/providers/prov_003/update", headers=auth(op),
|
||
|
|
json={"level": "premium", "status": "active"},
|
||
|
|
)
|
||
|
|
assert res.status_code == 200, res.text
|
||
|
|
assert res.json()["status"] == "active"
|
||
|
|
|
||
|
|
|
||
|
|
def test_operator_content_publish(client):
|
||
|
|
op = login(client, "pine")
|
||
|
|
res = client.post(
|
||
|
|
"/admin/content", headers=auth(op),
|
||
|
|
json={"type": "policy", "title": "新政策", "status": "draft"},
|
||
|
|
)
|
||
|
|
assert res.status_code == 200, res.text
|
||
|
|
cid = res.json()["id"]
|
||
|
|
res = client.post(
|
||
|
|
f"/admin/content/{cid}/status", headers=auth(op), json={"status": "published"},
|
||
|
|
)
|
||
|
|
assert res.status_code == 200
|
||
|
|
assert res.json()["status"] == "published"
|
||
|
|
|
||
|
|
|
||
|
|
def test_operator_config_update(client):
|
||
|
|
op = login(client, "pine")
|
||
|
|
res = client.put(
|
||
|
|
"/admin/config/platform.fee.task", headers=auth(op), json={"value": "0.06"},
|
||
|
|
)
|
||
|
|
assert res.status_code == 200, res.text
|
||
|
|
assert res.json()["value"] == "0.06"
|
||
|
|
|
||
|
|
|
||
|
|
def test_operator_stats_overview(client):
|
||
|
|
op = login(client, "pine")
|
||
|
|
stats = client.get("/admin/stats/overview", headers=auth(op)).json()
|
||
|
|
assert stats["user_count"] >= 7
|
||
|
|
assert stats["task_count"] >= 3
|
||
|
|
|
||
|
|
|
||
|
|
def test_non_operator_cannot_manage_tasks(client):
|
||
|
|
ent = login(client, "ent01")
|
||
|
|
assert client.get("/admin/tasks", headers=auth(ent)).status_code == 403
|