"""专家答疑：会员提问、只读本人、工作人员回复、待回复计数与超时。"""
from datetime import datetime, timedelta
import pytest
from fastapi.testclient import TestClient
from app.core.config import get_settings
from app.core.security import create_access_token
from app.models.expert import is_overdue, workday_deadline
from app.services.email_code import _key
from app.services.redis_client import get_redis
@pytest.fixture
def client(monkeypatch):
monkeypatch.setenv("JWT_SECRET", "test-secret-must-be-32-characters!")
monkeypatch.setenv("API_PAYLOAD_ENCRYPT", "false")
monkeypatch.setenv("SMTP_HOST", "")
get_settings.cache_clear()
if not get_settings().database_url.strip():
pytest.skip("DATABASE_URL 为空")
try:
get_redis().ping()
except Exception:
pytest.skip("Redis 不可达")
monkeypatch.setattr("app.services.email_code.send_email", lambda **kwargs: None)
from app.main import create_app
with TestClient(create_app()) as api:
yield api
get_settings.cache_clear()
def _minio_ready() -> bool:
try:
from app.services.storage import ensure_buckets
ensure_buckets()
return True
except Exception:
return False
def _code(purpose: str, email: str) -> str:
stored = get_redis().get(_key(purpose, email))
assert stored is not None
if isinstance(stored, bytes):
stored = stored.decode("utf-8")
return stored
def _register(client, email: str, password: str = "Passw0rd1") -> str:
assert client.post("/api/site/auth/email-code", json={"email": email, "purpose": "register"}).status_code == 200
code = _code("register", email)
registered = client.post(
"/api/site/auth/register",
json={"email": email, "code": code, "password": password, "privacyAccepted": True},
)
assert registered.status_code == 200, registered.text
return registered.json()["accessToken"]
def test_workday_deadline_skips_weekend():
# 周五提交，3 个工作日后是下周三
friday = datetime(2025, 1, 3, 10, 0, 0) # Friday
assert workday_deadline(friday, 3) == datetime(2025, 1, 8).date() # Wed
assert is_overdue(friday, 3, now=datetime(2025, 1, 8, 9, 0, 0)) is False
assert is_overdue(friday, 3, now=datetime(2025, 1, 9, 9, 0, 0)) is True
def test_expert_qa_ask_reply_isolation_and_pending_count(client):
if not _minio_ready():
pytest.skip("MinIO 不可达")
import time
email_a = f"qa-a-{int(time.time() * 1000)}@example.com"
email_b = f"qa-b-{int(time.time() * 1000)}@example.com"
token_a = _register(client, email_a)
token_b = _register(client, email_b)
headers_a = {"Authorization": f"Bearer {token_a}"}
headers_b = {"Authorization": f"Bearer {token_b}"}
staff = {"Authorization": f"Bearer {create_access_token('1', 'staff')}"}
denied = client.post(
"/api/site/member/expert-questions",
json={"title": "未登录", "body": "x", "attachmentFileIds": []},
)
assert denied.status_code in (401, 403)
uploaded = client.post(
"/api/site/member/files",
headers=headers_a,
files={"file": ("证据.pdf", b"%PDF-1.4\nqa", "application/pdf")},
)
assert uploaded.status_code == 200, uploaded.text
file_id = uploaded.json()["id"]
created = client.post(
"/api/site/member/expert-questions",
headers=headers_a,
json={
"title": "住宅估价流程咨询",
"body": "请问住宅估价需要哪些材料？",
"attachmentFileIds": [file_id],
},
)
assert created.status_code == 200, created.text
question_id = created.json()["id"]
mine = client.get("/api/site/member/expert-questions", headers=headers_a)
assert mine.status_code == 200, mine.text
assert mine.json()["items"][0]["title"] == "住宅估价流程咨询"
assert mine.json()["items"][0]["status"] == "pending"
other = client.get(f"/api/site/member/expert-questions/{question_id}", headers=headers_b)
assert other.status_code == 404
other_list = client.get("/api/site/member/expert-questions", headers=headers_b)
assert other_list.json()["items"] == []
pending = client.get("/api/admin/expert-questions/pending-count", headers=staff)
assert pending.status_code == 200, pending.text
assert pending.json()["count"] >= 1
listed = client.get("/api/admin/expert-questions?status=pending", headers=staff)
assert listed.status_code == 200, listed.text
assert any(item["id"] == question_id for item in listed.json()["items"])
member_reply = client.post(
f"/api/admin/expert-questions/{question_id}/reply",
headers=headers_a,
json={"bodyHtml": "会员不得回复
", "attachmentFileIds": []},
)
assert member_reply.status_code in (401, 403)
replied = client.post(
f"/api/admin/expert-questions/{question_id}/reply",
headers=staff,
json={
"bodyHtml": "请准备产权证与房屋平面图
",
"attachmentFileIds": [],
},
)
assert replied.status_code == 200, replied.text
detail = client.get(f"/api/site/member/expert-questions/{question_id}", headers=headers_a)
assert detail.status_code == 200, detail.text
assert detail.json()["status"] == "replied"
assert "产权证" in detail.json()["reply"]["bodyHtml"]
assert "script" not in detail.json()["reply"]["bodyHtml"].lower()
assert detail.json()["attachments"][0]["filename"] == "证据.pdf"
assert "contentType" in detail.json()["attachments"][0]
after = client.get("/api/admin/expert-questions/pending-count", headers=staff)
assert after.json()["count"] == pending.json()["count"] - 1
# 人为把另一条提问改成超时
overdue_created = client.post(
"/api/site/member/expert-questions",
headers=headers_a,
json={"title": "超时样例", "body": "等待回复", "attachmentFileIds": []},
)
assert overdue_created.status_code == 200, overdue_created.text
overdue_id = overdue_created.json()["id"]
from app.db import SessionLocal
from app.models.expert import ExpertQuestion
db = SessionLocal()
try:
row = db.get(ExpertQuestion, overdue_id)
row.created_at = datetime.now() - timedelta(days=14)
db.commit()
finally:
db.close()
admin_detail = client.get(f"/api/admin/expert-questions/{overdue_id}", headers=staff)
assert admin_detail.status_code == 200, admin_detail.text
assert admin_detail.json()["overdue"] is True