""" 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 行）：详情与取消只能操作本人预约。 断言必须是 **4xx**，不能只写 `!= 200`：后者会把 500 当成「越权被拒」放行。 此类服务端错误曾长期被这两处弱断言掩盖（`cancelAppointment` 抛裸 RuntimeException → 兜底 500），只有并发用例的 `< 500` 断言才咬了它一口。 """ 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 400 <= detail.status_code < 500, ( f"他人读取我的预约详情应被 4xx 拒绝（实际 HTTP {detail.status_code}）" ) cancel = api.call( "POST", f"/service-appointments/{created['id']}/cancel", token=stranger["token"], json={"reason": "越权取消"}, ) assert 400 <= cancel.status_code < 500, ( f"他人取消我的预约应被 4xx 拒绝（实际 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）， 与用户侧不校验状态形成对照 —— 仅断言**已实现**的那一侧。 断言 4xx 而非 `!= 200`：状态冲突应当是 409，不是 500。 """ 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 400 <= resp.status_code < 500, ( f"已完成预约不得被运营取消，且必须是 4xx（实际 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}" )