146 lines
4.9 KiB
Python
146 lines
4.9 KiB
Python
"""
|
|
E2E 测试：Phase 13 - 红娘服务转介
|
|
验证 INV-005（红娘不可代用户操作）及转介追踪链路
|
|
"""
|
|
import os
|
|
import pytest
|
|
import requests
|
|
|
|
BASE_URL = "http://localhost:8085/api/v1"
|
|
E2E_PASSWORD = os.environ["E2E_PASSWORD"]
|
|
E2E_USER_TOKEN = os.environ["E2E_USER_TOKEN"]
|
|
E2E_DB_PASSWORD = os.environ["E2E_DB_PASSWORD"]
|
|
|
|
# ---- 工具函数 ----
|
|
def get_user_token():
|
|
return E2E_USER_TOKEN
|
|
|
|
def get_matchmaker_token():
|
|
r = requests.post(f"{BASE_URL}/auth/operator/login", json={"username": "mm001", "password": E2E_PASSWORD})
|
|
return r.json()["data"]["accessToken"]
|
|
|
|
def auth_header(token):
|
|
return {"Authorization": f"Bearer {token}"}
|
|
|
|
|
|
# ---- P13-T05: 转介确认后激活 ----
|
|
def test_referral_requires_user_confirmation():
|
|
"""红娘转介后，预约状态为 submitted，需用户 confirm 后才变为 pending_confirmation"""
|
|
user_token = get_user_token()
|
|
mm_token = get_matchmaker_token()
|
|
|
|
# 红娘创建转介预约
|
|
r = requests.post(
|
|
f"{BASE_URL}/referrals/service-station",
|
|
headers=auth_header(mm_token),
|
|
json={"userId": "user-test-001", "locationId": "loc-001", "serviceType": "counselor_consultation"}
|
|
)
|
|
assert r.status_code == 200
|
|
appt_id = r.json()["data"]["appointmentId"]
|
|
assert r.json()["data"]["status"] == "submitted"
|
|
assert r.json()["data"]["needsUserConfirmation"] is True
|
|
|
|
# 用户确认
|
|
r2 = requests.post(
|
|
f"{BASE_URL}/referrals/{appt_id}/confirm",
|
|
headers=auth_header(user_token),
|
|
json={"targetType": "service_station"}
|
|
)
|
|
assert r2.status_code == 200
|
|
assert r2.json()["data"]["status"] == "pending_confirmation"
|
|
|
|
|
|
# ---- P13-T02: 转介活动 ----
|
|
def test_matchmaker_refers_activity():
|
|
"""红娘转介用户到公益活动，需用户确认"""
|
|
user_token = get_user_token()
|
|
mm_token = get_matchmaker_token()
|
|
|
|
r = requests.post(
|
|
f"{BASE_URL}/referrals/activity",
|
|
headers=auth_header(mm_token),
|
|
json={"userId": "user-test-001", "activityId": "act-001",
|
|
"emergencyContactName": "张某某", "emergencyContactPhoneEnc": "enc123"}
|
|
)
|
|
assert r.status_code == 200
|
|
reg_id = r.json()["data"]["registrationId"]
|
|
assert r.json()["data"]["referralType"] == "activity"
|
|
assert r.json()["data"]["needsUserConfirmation"] is True
|
|
|
|
# 用户确认
|
|
r2 = requests.post(
|
|
f"{BASE_URL}/referrals/{reg_id}/confirm",
|
|
headers=auth_header(user_token),
|
|
json={"targetType": "activity"}
|
|
)
|
|
assert r2.status_code == 200
|
|
|
|
|
|
# ---- P13-T04: 转介历史查询 ----
|
|
def test_referral_history():
|
|
"""查询用户的转介历史记录"""
|
|
user_token = get_user_token()
|
|
r = requests.get(f"{BASE_URL}/referrals/my-history", headers=auth_header(user_token))
|
|
assert r.status_code == 200
|
|
history = r.json()["data"]
|
|
assert isinstance(history, list)
|
|
|
|
|
|
# ---- INV-005: 红娘不能确认用户的转介 ----
|
|
def test_matchmaker_cannot_confirm_for_user():
|
|
"""INV-005: 红娘不能替用户确认转介，必须用户自己确认"""
|
|
mm_token = get_matchmaker_token()
|
|
|
|
# 红娘尝试确认（应该失败）
|
|
r = requests.post(
|
|
f"{BASE_URL}/referrals/some-id/confirm",
|
|
headers=auth_header(mm_token),
|
|
json={"targetType": "service_station"}
|
|
)
|
|
# 非对应用户ID，应返回 403 或 404
|
|
assert r.status_code in (400, 403, 404)
|
|
|
|
|
|
# ---- INV-005: 红娘不能取消用户的转介 ----
|
|
def test_matchmaker_cannot_cancel_for_user():
|
|
"""红娘不能取消用户的预约，只能用户自己取消"""
|
|
mm_token = get_matchmaker_token()
|
|
r = requests.post(
|
|
f"{BASE_URL}/service-appointments/some-id/cancel",
|
|
headers=auth_header(mm_token),
|
|
json={"reason": "测试取消"}
|
|
)
|
|
assert r.status_code in (400, 403, 404)
|
|
|
|
|
|
# ---- 转介来源追踪验证 ----
|
|
def test_referral_source_tracking():
|
|
"""验证转介时 source 字段正确设置"""
|
|
mm_token = get_matchmaker_token()
|
|
r = requests.post(
|
|
f"{BASE_URL}/referrals/service-station",
|
|
headers=auth_header(mm_token),
|
|
json={"userId": "user-test-001", "locationId": "loc-001", "serviceType": "station_visit"}
|
|
)
|
|
assert r.status_code == 200
|
|
appt_id = r.json()["data"]["appointmentId"]
|
|
|
|
# 查询预约详情，验证 source
|
|
r2 = requests.get(f"{BASE_URL}/service-appointments/{appt_id}",
|
|
headers=auth_header("Bearer dummy"))
|
|
# 通过数据库直接验证 source 字段
|
|
import psycopg2
|
|
conn = psycopg2.connect(f"dbname=danyang_match user=postgres password={E2E_DB_PASSWORD} host=localhost port=5434")
|
|
cur = conn.cursor()
|
|
cur.execute("SELECT source, request_source, created_by_operator_id FROM service_appointments WHERE id = %s", (appt_id,))
|
|
row = cur.fetchone()
|
|
assert row[0] == "matchmaker_referral"
|
|
assert row[1] == "matchmaker_referral"
|
|
assert row[2] is not None
|
|
cur.close()
|
|
conn.close()
|
|
|
|
|
|
if __name__ == "__main__":
|
|
pytest.main([__file__, "-v"])
|