danyang-match/tests/e2e/counseling_v11_test.py

"""
E2E 测试：Phase 13A - 公益婚恋咨询与婚前辅导
验证 INV-012（Counselor assigned scope）等不变量
"""
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_counselor_token():
    r = requests.post(f"{BASE_URL}/auth/operator/login", json={"username": "counselor01", "password": E2E_PASSWORD})
    return r.json()["data"]["accessToken"]

def auth_header(token):
    return {"Authorization": f"Bearer {token}"}


# ---- P13A-T05: 获取咨询介绍 ----
def test_get_counseling_intro():
    r = requests.get(f"{BASE_URL}/counseling/services/intro")
    assert r.status_code == 200
    data = r.json()["data"]
    assert isinstance(data, list)
    assert len(data) >= 3
    types = {item["type"] for item in data}
    assert "relationship_guidance" in types
    assert "premarital_guidance" in types


# ---- P13A-T06: 申请咨询 ----
def test_create_counseling():
    token = get_user_token()
    r = requests.post(
        f"{BASE_URL}/counseling/services",
        headers=auth_header(token),
        json={"serviceType": "relationship_guidance", "userRequestSummary": "恋爱沟通困难"}
    )
    assert r.status_code == 200
    data = r.json()["data"]
    assert data["status"] == "pending"
    assert data["serviceType"] == "relationship_guidance"
    assert data["requestSource"] == "user_self"
    counseling_id = data["id"]

    # 验证通知已发送（通过查询数据库）
    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 COUNT(*) FROM notifications WHERE business_id = %s AND type = 'counseling'", (counseling_id,))
    count = cur.fetchone()[0]
    assert count >= 1
    cur.close()
    conn.close()


# ---- P13A-T07: 获取我的咨询列表 ----
def test_get_my_counselings():
    token = get_user_token()
    r = requests.get(f"{BASE_URL}/counseling/services", headers=auth_header(token))
    assert r.status_code == 200
    assert isinstance(r.json()["data"], list)


# ---- P13A-T08: 咨询详情（无权访问他人记录） ----
def test_counseling_detail_no_access():
    token = get_user_token()
    # 尝试访问不属于自己的咨询
    r = requests.get(f"{BASE_URL}/counseling/services/non-existent-id", headers=auth_header(token))
    assert r.status_code in (400, 404)


# ---- P13A-T09: 取消咨询 ----
def test_cancel_counseling():
    token = get_user_token()
    # 先创建再取消
    create_r = requests.post(
        f"{BASE_URL}/counseling/services",
        headers=auth_header(token),
        json={"serviceType": "emotional_support", "userRequestSummary": "测试取消"}
    )
    counseling_id = create_r.json()["data"]["id"]
    cancel_r = requests.post(
        f"{BASE_URL}/counseling/services/{counseling_id}/cancel",
        headers=auth_header(token)
    )
    assert cancel_r.status_code == 200
    assert cancel_r.json()["data"]["status"] == "cancelled"


# ---- INV-012: Counselor只能访问分配给自己的服务 ----
def test_counselor_assigned_scope():
    counselor_token = get_counselor_token()
    # Counselor 只能看到分配给自己的
    r = requests.get(f"{BASE_URL}/counseling/services/operator/assigned", headers=auth_header(counselor_token))
    assert r.status_code == 200
    # 不应包含未分配给自己的记录
    # （假设数据库中 counselor01 没有被分配的记录，返回空列表）
    assert isinstance(r.json()["data"], list)


# ---- INV-013: Matchmaker不能读取Counseling正文 ----
# Phase 23 KEEP-VERIFY（2026-09-18）：原 test_matchmaker_cannot_read_counseling_body
# 的函数体为 `pass` 空壳（注释自述「在当前测试框架下简化验证」），从未真实断言过任何权限，
# 却在 docs/STATUS.md 中被计为 P23-T06「咨询 E2E ✅」的一部分。
# 计划 Phase 23 Gate「不允许 skip/pass 空壳」→ 已移除该空壳。
#
# INV-013 现由后端真实行为用例覆盖：
#   backend/.../invariants/V11InvariantsTest#inv013MatchmakerCannotReadCounselingBody
#   走真实 CounselingService.getDetail：断言 Matchmaker 读到的 user_request_summary 为 null，
#   并以「assigned Counselor 能读到正文」作反证，避免全局脱敏造成的假阳性；
#   已通过负向注入验证（把 isMatchmaker 分支改为返回全文 → 该用例精准变红）。
#
# E2E 层待补场景：用真实红娘 token 调用咨询详情接口并断言被拒。


if __name__ == "__main__":
    pytest.main([__file__, "-v"])