320 lines
12 KiB
Python
320 lines
12 KiB
Python
"""行业自律：三类栏目文章公开阅读；投诉加密保存，查询码只看状态。"""
|
|
|
|
import base64
|
|
import hashlib
|
|
from datetime import datetime, timedelta
|
|
|
|
import pytest
|
|
from fastapi.testclient import TestClient
|
|
from sqlalchemy import select
|
|
|
|
from app.core.config import get_settings
|
|
from app.core.security import create_access_token
|
|
|
|
|
|
@pytest.fixture
|
|
def client(monkeypatch):
|
|
monkeypatch.setenv("JWT_SECRET", "test-secret-must-be-32-characters!")
|
|
monkeypatch.setenv("API_PAYLOAD_ENCRYPT", "false")
|
|
key = base64.b64encode(hashlib.sha256(b"discipline-field-key").digest()).decode("ascii")
|
|
monkeypatch.setenv("FIELD_AES_KEY", key)
|
|
get_settings.cache_clear()
|
|
if not get_settings().database_url.strip():
|
|
pytest.skip("DATABASE_URL 为空")
|
|
from app.main import create_app
|
|
|
|
with TestClient(create_app()) as api:
|
|
yield api
|
|
get_settings.cache_clear()
|
|
|
|
|
|
def _headers(token_type: str) -> dict[str, str]:
|
|
return {"Authorization": f"Bearer {create_access_token('1', token_type)}"}
|
|
|
|
|
|
def _minio_ready() -> bool:
|
|
try:
|
|
from app.services.storage import ensure_buckets
|
|
|
|
ensure_buckets()
|
|
return True
|
|
except Exception:
|
|
return False
|
|
|
|
|
|
def _png() -> bytes:
|
|
return (
|
|
b"\x89PNG\r\n\x1a\n\x00\x00\x00\rIHDR\x00\x00\x00\x01\x00\x00\x00\x01"
|
|
b"\x08\x02\x00\x00\x00\x90wS\xde\x00\x00\x00\x0cIDATx\x9cc\xf8\x0f"
|
|
b"\x00\x00\x01\x01\x00\x05\x18\xd8N\x00\x00\x00\x00IEND\xaeB`\x82"
|
|
)
|
|
|
|
|
|
def _clear_category(client, staff: dict[str, str], category: str) -> None:
|
|
listed = client.get(f"/api/admin/about/articles/{category}", headers=staff)
|
|
assert listed.status_code == 200, listed.text
|
|
for item in listed.json().get("items") or []:
|
|
deleted = client.delete(f"/api/admin/home/articles/{item['id']}", headers=staff)
|
|
assert deleted.status_code in (200, 204), deleted.text
|
|
|
|
|
|
def _purge_complaints(codes: list[str]) -> list[int]:
|
|
from app.db import SessionLocal
|
|
from app.models.discipline import DisciplineComplaint, DisciplineComplaintFile
|
|
|
|
db = SessionLocal()
|
|
file_ids: list[int] = []
|
|
try:
|
|
rows = db.scalars(select(DisciplineComplaint).where(DisciplineComplaint.code.in_(codes))).all()
|
|
ids = [row.id for row in rows]
|
|
if not ids:
|
|
return []
|
|
links = db.scalars(
|
|
select(DisciplineComplaintFile).where(DisciplineComplaintFile.complaint_id.in_(ids))
|
|
).all()
|
|
file_ids = [link.file_id for link in links]
|
|
for link in links:
|
|
db.delete(link)
|
|
for row in rows:
|
|
db.delete(row)
|
|
db.commit()
|
|
finally:
|
|
db.close()
|
|
return file_ids
|
|
|
|
|
|
def test_discipline_articles_notice_and_complaint(client):
|
|
if not _minio_ready():
|
|
pytest.skip("MinIO 不可达")
|
|
staff = _headers("staff")
|
|
member = _headers("member")
|
|
article_ids: list[int] = []
|
|
file_ids: list[int] = []
|
|
codes: list[str] = []
|
|
word_id = None
|
|
previous_workdays = None
|
|
try:
|
|
for category, placeholder in (
|
|
("warning_cases", "暂无警示案例"),
|
|
("integrity_records", "暂无诚信档案"),
|
|
("discipline_results", "暂无结果公示"),
|
|
):
|
|
_clear_category(client, staff, category)
|
|
empty = client.get(f"/api/site/about/articles/{category}")
|
|
assert empty.status_code == 200, empty.text
|
|
assert placeholder in empty.text
|
|
|
|
denied = client.get("/api/admin/discipline/complaints", headers=member)
|
|
assert denied.status_code == 403, denied.text
|
|
assert client.get("/api/admin/discipline/complaints").status_code == 401
|
|
|
|
cover = client.post(
|
|
"/api/admin/files",
|
|
headers=staff,
|
|
data={"bucket": "public"},
|
|
files={"file": ("cover.png", _png(), "image/png")},
|
|
)
|
|
assert cover.status_code == 200, cover.text
|
|
file_ids.append(cover.json()["id"])
|
|
attachment = client.post(
|
|
"/api/admin/files",
|
|
headers=staff,
|
|
data={"bucket": "public"},
|
|
files={"file": ("note.txt", b"public-note", "text/plain")},
|
|
)
|
|
assert attachment.status_code == 200, attachment.text
|
|
file_ids.append(attachment.json()["id"])
|
|
|
|
draft = client.post(
|
|
"/api/admin/home/articles",
|
|
headers=staff,
|
|
json={
|
|
"category": "warning_cases",
|
|
"title": "未发布警示",
|
|
"published": False,
|
|
"bodyHtml": "<p>内部草稿</p>",
|
|
"attachmentFileIds": [],
|
|
},
|
|
)
|
|
assert draft.status_code == 200, draft.text
|
|
article_ids.append(draft.json()["id"])
|
|
published = client.post(
|
|
"/api/admin/home/articles",
|
|
headers=staff,
|
|
json={
|
|
"category": "discipline_results",
|
|
"title": "某机构处理结果",
|
|
"published": True,
|
|
"bodyHtml": "<p>已脱敏的处理结果</p><script>alert(1)</script>",
|
|
"imageFileId": file_ids[0],
|
|
"attachmentFileIds": [file_ids[1]],
|
|
},
|
|
)
|
|
assert published.status_code == 200, published.text
|
|
article_ids.append(published.json()["id"])
|
|
|
|
hidden = client.get("/api/site/about/articles/warning_cases")
|
|
assert "未发布警示" not in hidden.text
|
|
shown = client.get("/api/site/about/articles/discipline_results")
|
|
titles = [item["title"] for item in shown.json()["items"]]
|
|
assert "某机构处理结果" in titles
|
|
detail = client.get(f"/api/site/home/articles/{published.json()['id']}")
|
|
assert "已脱敏的处理结果" in detail.text
|
|
assert "script" not in detail.text
|
|
assert "note.txt" in detail.text
|
|
|
|
home = client.get("/api/admin/home", headers=staff)
|
|
categories = {item["category"] for item in home.json()["articles"]}
|
|
assert "warning_cases" not in categories
|
|
assert "discipline_results" not in categories
|
|
|
|
settings = client.get("/api/admin/site-settings", headers=staff)
|
|
assert settings.status_code == 200, settings.text
|
|
previous_workdays = settings.json().get("complaint_workdays", "")
|
|
updated = client.put(
|
|
"/api/admin/site-settings",
|
|
headers=staff,
|
|
json={"complaint_workdays": 12},
|
|
)
|
|
assert updated.status_code == 200, updated.text
|
|
|
|
notice = client.put(
|
|
"/api/admin/discipline/notice",
|
|
headers=staff,
|
|
json={
|
|
"phone": "0931-1234567",
|
|
"email": "report@example.com",
|
|
"scopeHtml": "<p>受理估价执业违规</p><script>alert(1)</script>",
|
|
"processHtml": "<p>受理后核查</p>",
|
|
},
|
|
)
|
|
assert notice.status_code == 200, notice.text
|
|
assert notice.json()["workdays"] == 12
|
|
public_notice = client.get("/api/site/discipline/notice")
|
|
assert public_notice.status_code == 200, public_notice.text
|
|
assert "0931-1234567" in public_notice.text
|
|
assert "受理估价执业违规" in public_notice.text
|
|
assert "script" not in public_notice.text
|
|
assert public_notice.json()["workdays"] == 12
|
|
|
|
missing = client.post(
|
|
"/api/site/discipline/complaints",
|
|
data={
|
|
"title": "缺证据",
|
|
"subject": "某机构",
|
|
"body": "没有附件",
|
|
"contactName": "李四",
|
|
"mobile": "13800138000",
|
|
},
|
|
)
|
|
assert missing.status_code == 400, missing.text
|
|
assert "证据" in missing.text
|
|
|
|
word = "自律禁词" + hashlib.sha256(b"discipline-word").hexdigest()[:8]
|
|
created_word = client.post("/api/admin/sensitive-words", headers=staff, json={"word": word})
|
|
assert created_word.status_code == 200, created_word.text
|
|
word_id = created_word.json()["id"]
|
|
blocked = client.post(
|
|
"/api/site/discipline/complaints",
|
|
data={
|
|
"title": f"标题含{word}",
|
|
"subject": "某机构",
|
|
"body": "事实说明",
|
|
"contactName": "李四",
|
|
"mobile": "13800138000",
|
|
},
|
|
files=[("evidence", ("proof.png", _png(), "image/png"))],
|
|
)
|
|
assert blocked.status_code == 400, blocked.text
|
|
assert "敏感词" in blocked.text
|
|
|
|
secret = "只应工作人员看到的违规事实"
|
|
submitted = client.post(
|
|
"/api/site/discipline/complaints",
|
|
data={
|
|
"title": "估价报告违规",
|
|
"subject": "某估价机构",
|
|
"body": secret,
|
|
"contactName": "张三",
|
|
"mobile": "138-0013-8000",
|
|
"email": "a@example.com",
|
|
},
|
|
files=[("evidence", ("proof.png", _png(), "image/png"))],
|
|
)
|
|
assert submitted.status_code == 200, submitted.text
|
|
payload = submitted.json()
|
|
codes.append(payload["code"])
|
|
assert payload["status"] == "pending"
|
|
assert payload["workdays"] == 12
|
|
assert secret not in submitted.text
|
|
assert "13800138000" not in submitted.text
|
|
|
|
listed = client.get("/api/admin/discipline/complaints", headers=staff)
|
|
assert listed.status_code == 200, listed.text
|
|
item = next(row for row in listed.json()["items"] if row["code"] == payload["code"])
|
|
assert item["mobileMask"] == "138****8000"
|
|
assert item["emailMask"] == "a***@example.com"
|
|
assert secret not in listed.text
|
|
assert "13800138000" not in listed.text
|
|
assert item["attachmentCount"] == 1
|
|
|
|
opened = client.get(f"/api/admin/discipline/complaints/{item['id']}", headers=staff)
|
|
assert opened.status_code == 200, opened.text
|
|
assert opened.json()["mobile"] == "13800138000"
|
|
assert opened.json()["email"] == "a@example.com"
|
|
assert secret in opened.text
|
|
assert opened.json()["attachments"][0]["filename"] == "proof.png"
|
|
assert opened.json()["attachments"][0]["url"]
|
|
file_ids.append(opened.json()["attachments"][0]["id"])
|
|
|
|
status = client.get(f"/api/site/discipline/complaints/{payload['code']}")
|
|
assert status.status_code == 200, status.text
|
|
assert status.json()["statusLabel"] == "待处理"
|
|
assert secret not in status.text
|
|
assert "13800138000" not in status.text
|
|
assert "张三" not in status.text
|
|
assert "proof.png" not in status.text
|
|
|
|
from app.db import SessionLocal
|
|
from app.models.discipline import DisciplineComplaint
|
|
|
|
db = SessionLocal()
|
|
try:
|
|
row = db.scalar(select(DisciplineComplaint).where(DisciplineComplaint.code == payload["code"]))
|
|
assert row is not None
|
|
row.created_at = datetime.now() - timedelta(days=40)
|
|
db.commit()
|
|
finally:
|
|
db.close()
|
|
overdue = client.get("/api/admin/discipline/complaints?status=pending", headers=staff)
|
|
overdue_item = next(row for row in overdue.json()["items"] if row["code"] == payload["code"])
|
|
assert overdue_item["overdue"] is True
|
|
public_overdue = client.get(f"/api/site/discipline/complaints/{payload['code']}")
|
|
assert public_overdue.json()["overdue"] is True
|
|
|
|
done = client.post(f"/api/admin/discipline/complaints/{item['id']}/process", headers=staff)
|
|
assert done.status_code == 200, done.text
|
|
finished = client.get(f"/api/site/discipline/complaints/{payload['code'].lower()}")
|
|
assert finished.json()["status"] == "processed"
|
|
assert finished.json()["overdue"] is False
|
|
assert client.get("/api/site/discipline/complaints/NO-SUCH").status_code == 404
|
|
finally:
|
|
for article_id in article_ids:
|
|
client.delete(f"/api/admin/home/articles/{article_id}", headers=staff)
|
|
evidence_ids = _purge_complaints(codes)
|
|
for file_id in [*file_ids, *evidence_ids]:
|
|
client.delete(f"/api/admin/files/{file_id}", headers=staff)
|
|
if word_id is not None:
|
|
client.delete(f"/api/admin/sensitive-words/{word_id}", headers=staff)
|
|
client.put(
|
|
"/api/admin/discipline/notice",
|
|
headers=staff,
|
|
json={"phone": "", "email": "", "scopeHtml": "", "processHtml": ""},
|
|
)
|
|
if previous_workdays is not None:
|
|
client.put(
|
|
"/api/admin/site-settings",
|
|
headers=staff,
|
|
json={"complaint_workdays": previous_workdays},
|
|
)
|