danyang-match/tests/e2e/service_appointment_test.py

"""
V1.1 服务预约 E2E —— 场景真实实现（Phase 23 Task 2.1，2026-09-19）

原文件 7 个测试函数**全部**为 `pass` 空壳（基线见 git b46b48e），依赖不存在的
`service_location_factory` / `appointment_factory` / `operator_client` 等 fixture，
从未执行过，却在 `docs/STATUS.md` 中被计为 P23-T08「服务预约 E2E ✅」。
本文件按文件头保留的场景清单逐条实现为**真实 HTTP 用例**。

实现纪律（与 conftest 一致，不得回退）
------------------------------------
1. 不 mock 业务行为：走真实 HTTP + 真实数据库；
2. 每个用例使用**独立新用户**，避免用例互相污染；
3. 栈不可达时由 conftest `pytest.exit(returncode=4)`，**不退化为 skip**。

被验证的调用链（实测自 ServiceAppointmentController / ServiceAppointmentService）
--------------------------------------------------------------------------------
    POST /service-appointments                        {serviceLocationId, serviceType, preferredDate, userNote}
    GET  /service-appointments                        我的预约列表
    GET  /service-appointments/{id}                   详情（仅本人可见）
    POST /service-appointments/{id}/cancel            {reason}   本人取消
    POST /service-appointments/operator/{id}/confirm  运营确认（service_appointment.confirm）
    POST /service-appointments/operator/{id}/complete 运营完成（service_appointment.complete）
    POST /service-appointments/operator/{id}/cancel   运营取消（service_appointment.confirm）

关于场景 7（原 `test_cancel_after_confirmation_denied`）
-------------------------------------------------------
原空壳里的用例名是「已确认状态不可自行取消」，但**全仓没有任何规格支撑该规则**：
需求基线里用户侧取消的权威口径只有归属隔离 ——
「修复预约 IDOR：普通用户创建、详情、列表、取消只能操作本人」（V1.1 计划第 488 行）。
实测实现也确实**不按状态拦截用户取消**（`ServiceAppointmentService.cancelAppointment`
只校验 `userId` 归属）。Global Constraints 明令「不修改业务规则」，
因此本文件不把无规格支撑的期待写成断言，而是实现**有规格支撑**的那一条：
归属隔离 + 运营侧的状态闸门（`cancelAppointmentByOperator` 明确拒绝
completed / cancelled 的预约）。原用例名的意图差异已单独上报，待产品裁定。

已知不对称（**本文件不固化、留待裁定**）
--------------------------------------
用户侧 `cancelAppointment` 不校验状态，因此**已完成**的预约也能被本人置为 cancelled；
而运营侧 `cancelAppointmentByOperator` 明确拒绝 completed。两侧口径不一致。
本文件不对此断言（会固化一个可疑行为），以「观察到的实现」形式在报告中说明。
"""
from __future__ import annotations

import os
import subprocess
from concurrent.futures import ThreadPoolExecutor
from datetime import date, timedelta
from typing import Any, Dict, Optional

import pytest

pytestmark = pytest.mark.e2e

SERVICE_TYPE = "counselor_consultation"
IDEMPOTENCY_HEADER = "Idempotency-Key"

# 预约日期取「今天 + 30 天」，避免用例随日历漂移
PREFERRED_DATE = (date.today() + timedelta(days=30)).isoformat()

# 与 tests/e2e/sql/seed_e2e_service_station.sql 中的固定 UUID 一一对应
E2E_STATION_ID = "e2e00000-0000-0000-0000-0000000000b1"
# 该站点在 seed 中关联的运营（service_location_operators），
# 也就是唯一能确认/完成本用例预约的运营账号（数据范围约束，super_admin 不豁免）
STATION_OPERATOR = "e2e_station"

PG_CONTAINER = os.environ.get("E2E_PG_CONTAINER", "danyang-match-postgres")
PG_DB = os.environ.get("E2E_PG_DB", "danyang_match")

_station_ready = False


def _ensure_station_seed() -> None:
    """幂等确保「E2E 服务站 + 站点运营关联」存在。

    两样都必需：新库无站点；且运营确认受数据范围约束
    （`assertOperatorScope`，super_admin 也不豁免），没有关联必然 403。
    """
    global _station_ready
    if _station_ready:
        return
    seed_path = os.path.join(os.path.dirname(__file__), "sql", "seed_e2e_service_station.sql")
    with open(seed_path, encoding="utf-8") as fh:
        seed = fh.read()
    proc = subprocess.run(
        ["docker", "exec", "-i", PG_CONTAINER, "psql", "-U", "postgres", "-d", PG_DB,
         "-v", "ON_ERROR_STOP=1", "-q"],
        input=seed, capture_output=True, text=True,
    )
    assert proc.returncode == 0, (
        f"服务站前置数据 seed 失败（需要容器 {PG_CONTAINER} 可访问）：\n"
        f"stdout={proc.stdout}\nstderr={proc.stderr}"
    )
    _station_ready = True



# ------------------------------------------------------------------ 辅助


def _headers(key: Optional[str] = None) -> Dict[str, str]:
    return {IDEMPOTENCY_HEADER: key} if key else {}


def _station_id(api: Any, token: str) -> str:
    """确认 seed 出来的 E2E 服务站真实存在且可见，返回其 id。

    不用「取第一个 active 站点」：本机可能还有历史站点，
    而运营确认受数据范围约束 —— 必须是 seed 里关联过运营的那一个。
    """
    _ensure_station_seed()
    locations = api.ok("GET", "/operator/service-locations", token=token) or []
    ids = {loc.get("id") for loc in locations}
    assert E2E_STATION_ID in ids, (
        f"seed 的 E2E 服务站未出现在运营站点列表中（实得 {sorted(ids)}）"
    )
    return E2E_STATION_ID


def _create(api: Any, token: str, station_id: str, *, key: Optional[str] = None) -> Dict[str, Any]:
    return api.ok(
        "POST",
        "/service-appointments",
        token=token,
        headers=_headers(key),
        json={
            "serviceLocationId": station_id,
            "serviceType": SERVICE_TYPE,
            "preferredDate": PREFERRED_DATE,
            "userNote": "E2E 预约用例",
        },
    )


def _my_appointments(api: Any, token: str) -> list:
    return api.ok("GET", "/service-appointments", token=token) or []


def _by_id(api: Any, token: str, appointment_id: str) -> Optional[Dict[str, Any]]:
    for item in _my_appointments(api, token):
        if item.get("id") == appointment_id:
            return item
    return None


def _operator_confirm(api: Any, token: str, appointment_id: str) -> Dict[str, Any]:
    return api.ok("POST", f"/service-appointments/operator/{appointment_id}/confirm", token=token)


def _operator_complete(api: Any, token: str, appointment_id: str) -> Dict[str, Any]:
    return api.ok("POST", f"/service-appointments/operator/{appointment_id}/complete", token=token)


def _user_cancel(api: Any, token: str, appointment_id: str) -> Dict[str, Any]:
    return api.ok(
        "POST",
        f"/service-appointments/{appointment_id}/cancel",
        token=token,
        json={"reason": "E2E 取消"},
    )


@pytest.fixture
def station_id(api, operator_token) -> str:
    """seed 出来的 E2E 服务站 id（该站点已关联 STATION_OPERATOR）。"""
    return _station_id(api, operator_token(STATION_OPERATOR))


# ------------------------------------------------------------------ 场景 1


def test_submit_appointment(api, new_user, station_id):
    """用户可提交服务预约，初始状态为 submitted。"""
    user = new_user()

    created = _create(api, user["token"], station_id)

    assert created["id"], f"创建预约必须返回 id: {created}"
    assert created["status"] == "submitted", f"新建预约状态应为 submitted: {created}"
    assert created["serviceLocationId"] == station_id, f"服务站应回显: {created}"


# ------------------------------------------------------------------ 场景 2


def test_appointment_idempotent(api, new_user, station_id):
    """同一把 Idempotency-Key 的重复投递必须重放首次结果，不得产生第二条预约。"""
    user = new_user()
    key = f"e2e-appt-{user['code']}"

    first = _create(api, user["token"], station_id, key=key)
    second = _create(api, user["token"], station_id, key=key)

    assert second["id"] == first["id"], (
        f"同一幂等键必须重放同一预约（first={first['id']}, second={second['id']}）"
    )
    matching = [a for a in _my_appointments(api, user["token"]) if a.get("id") == first["id"]]
    assert len(matching) == 1, f"幂等键重复投递不得产生第二条预约（实得 {len(matching)} 条）"


# ------------------------------------------------------------------ 场景 3


def test_operator_can_confirm(api, new_user, operator_token, station_id):
    """后台可确认预约：submitted → confirmed。

    必须用**该站点关联的运营**（数据范围约束）：super_admin 也拿不到别站预约。
    """
    user = new_user()
    created = _create(api, user["token"], station_id)

    confirmed = _operator_confirm(api, operator_token(STATION_OPERATOR), created["id"])

    assert confirmed["status"] == "confirmed", f"运营确认后状态应为 confirmed: {confirmed}"
    mine = _by_id(api, user["token"], created["id"])
    assert mine is not None and mine["status"] == "confirmed", (
        f"用户侧应看到同一状态（不是运营侧自嗨）: {mine}"
    )


# ------------------------------------------------------------------ 场景 4


def test_appointment_concurrent_safety(api, new_user, operator_token, station_id):
    """同一预约「用户取消」与「运营确认」并发：不得出现 5xx，最终状态必须是二者之一。

    验收口径是**一致性**：无论谁赢，库里都只能停在 confirmed 或 cancelled 之一，
    不能两个都发生、也不能报 500。
    """
    user = new_user()
    created = _create(api, user["token"], station_id)
    appointment_id = created["id"]
    op_token = operator_token(STATION_OPERATOR)

    def do_cancel() -> int:
        return api.call(
            "POST", f"/service-appointments/{appointment_id}/cancel",
            token=user["token"], json={"reason": "并发取消"},
        ).status_code

    def do_confirm() -> int:
        return api.call(
            "POST", f"/service-appointments/operator/{appointment_id}/confirm",
            token=op_token,
        ).status_code

    with ThreadPoolExecutor(max_workers=2) as pool:
        cancel_code = pool.submit(do_cancel)
        confirm_code = pool.submit(do_confirm)
        codes = [cancel_code.result(), confirm_code.result()]

    assert all(code < 500 for code in codes), (
        f"并发取消/确认不得出现服务端错误（实际状态码 {codes}）"
    )

    final = _by_id(api, user["token"], appointment_id)
    assert final is not None, "并发后预约必须仍可读"
    assert final["status"] in {"confirmed", "cancelled"}, (
        f"并发后最终状态必须收敛到 confirmed 或 cancelled，实得 {final['status']}"
    )


# ------------------------------------------------------------------ 场景 5


def test_my_appointments_list(api, new_user, station_id):
    """我的预约列表只包含本人预约。"""
    owner = new_user()
    stranger = new_user()
    created = _create(api, owner["token"], station_id)

    mine = _my_appointments(api, owner["token"])
    assert any(a.get("id") == created["id"] for a in mine), (
        f"本人列表必须包含刚创建的预约: {mine}"
    )

    strangers = _my_appointments(api, stranger["token"])
    assert all(a.get("id") != created["id"] for a in strangers), (
        "预约不得出现在他人列表中（数据范围隔离）"
    )


# ------------------------------------------------------------------ 场景 6


def test_cancel_before_confirmation(api, new_user, station_id):
    """待确认（submitted）状态可取消。"""
    user = new_user()
    created = _create(api, user["token"], station_id)

    cancelled = _user_cancel(api, user["token"], created["id"])

    assert cancelled["status"] == "cancelled", f"取消后状态应为 cancelled: {cancelled}"


# ------------------------------------------------------------------ 场景 7（归属隔离 + 运营侧状态闸门）


def test_owner_scope_on_detail_and_cancel(api, new_user, station_id):
    """归属隔离（V1.1 计划第 488 行）：详情与取消只能操作本人预约。"""
    owner = new_user()
    stranger = new_user()
    created = _create(api, owner["token"], station_id)

    detail = api.call("GET", f"/service-appointments/{created['id']}", token=stranger["token"])
    assert detail.status_code != 200, (
        f"他人不得读取我的预约详情（实际 HTTP {detail.status_code}）"
    )

    cancel = api.call(
        "POST", f"/service-appointments/{created['id']}/cancel",
        token=stranger["token"], json={"reason": "越权取消"},
    )
    assert cancel.status_code != 200, (
        f"他人不得取消我的预约（实际 HTTP {cancel.status_code}）"
    )

    still = _by_id(api, owner["token"], created["id"])
    assert still is not None and still["status"] == "submitted", (
        f"越权取消失败后，本人预约状态不得被改动: {still}"
    )


def test_operator_cannot_cancel_completed(api, new_user, operator_token, station_id):
    """运营侧状态闸门：已完成的预约不得再被取消。

    这是 `cancelAppointmentByOperator` 明确实现的规则（拒绝 completed / cancelled），
    与用户侧不校验状态形成对照 —— 仅断言**已实现**的那一侧。
    """
    user = new_user()
    op_token = operator_token(STATION_OPERATOR)
    created = _create(api, user["token"], station_id)

    _operator_confirm(api, op_token, created["id"])
    completed = _operator_complete(api, op_token, created["id"])
    assert completed["status"] == "completed", f"前置条件失败：应可完成预约: {completed}"

    resp = api.call(
        "POST", f"/service-appointments/operator/{created['id']}/cancel",
        token=op_token, json={"reason": "试图取消已完成预约"},
    )
    assert resp.status_code != 200, (
        f"已完成预约不得被运营取消（实际 HTTP {resp.status_code}: {resp.text[:200]}）"
    )

    final = _by_id(api, user["token"], created["id"])
    assert final is not None and final["status"] == "completed", (
        f"被拒绝的取消不得改动状态: {final}"
    )