danyang-match/tests/e2e/conftest.py

"""
善缘丹阳 E2E 测试基建（pytest conftest）。

设计纪律（Phase 23 复验后确立，不得回退）：

1. **栈不可达 = 退出，不是跳过。** `skip` 会把"未执行"伪装成"通过"，
   这正是 Phase 23 清掉的 44 个 `pass` 空壳的另一种变体。因此本文件在
   活栈不可达时调用 `pytest.exit(returncode=4)` —— 明确报"没跑成"。
2. **不 mock 业务行为。** 所有 fixture 走真实 HTTP 与真实数据库；
   唯一被替换的是微信登录（后端 `MockWechatLoginProvider` 默认激活）。
3. **用户隔离。** 每个用例拿独立新用户（mock 登录同 code 同 openid，
   换 code 即换人），避免用例互相污染与"上一个用例造的数据让本用例变绿"。

环境变量：
- `E2E_BASE_URL`        活栈地址，默认 http://localhost:8085/api/v1
- `E2E_OPERATOR_PASSWORD` 运营账号密码（必需；CI 由生成的 .env 提供）
"""
from __future__ import annotations

import os
import time
import uuid
from typing import Any, Callable, Dict, Optional

import pytest
import requests

DEFAULT_BASE_URL = "http://localhost:8085/api/v1"
HEALTH_PATH = "/health/live"


class ApiClient:
    """极薄 HTTP 客户端：只负责超时与错误上下文，不做业务判断。"""

    def __init__(self, base_url: str, timeout: float = 15.0) -> None:
        self.base_url = base_url.rstrip("/")
        self.timeout = timeout
        self.session = requests.Session()
        # E2E 打的是 localhost 活栈。若宿主机设置了 http_proxy，
        # requests 默认会把 localhost 请求也发给代理并返回 502，
        # 表现为"连接失败"而非真实的业务结果——直接绕开环境变量。
        self.session.trust_env = False

    def call(
        self,
        method: str,
        path: str,
        *,
        token: Optional[str] = None,
        json: Optional[Dict[str, Any]] = None,
        params: Optional[Dict[str, Any]] = None,
        headers: Optional[Dict[str, str]] = None,
    ) -> requests.Response:
        merged = {"Authorization": f"Bearer {token}"} if token else {}
        if headers:
            # 显式传入的头覆盖默认头（幂等键等按用例定制）
            merged.update(headers)
        return self.session.request(
            method,
            f"{self.base_url}{path}",
            headers=merged,
            json=json,
            params=params,
            timeout=self.timeout,
        )

    def ok(self, method: str, path: str, **kwargs: Any) -> Any:
        """断言 HTTP 200 且业务 `success=true`，返回 `data`。"""
        resp = self.call(method, path, **kwargs)
        assert resp.status_code == 200, (
            f"{method} {path} -> HTTP {resp.status_code}: {resp.text[:400]}"
        )
        body = resp.json()
        assert body.get("success") is True, f"{method} {path} -> 业务失败: {body}"
        return body.get("data")


# ---------------------------------------------------------------- fixtures


@pytest.fixture(scope="session")
def base_url() -> str:
    return os.environ.get("E2E_BASE_URL", DEFAULT_BASE_URL).rstrip("/")


@pytest.fixture(scope="session")
def api(base_url: str) -> ApiClient:
    return ApiClient(base_url)


@pytest.fixture(scope="session", autouse=True)
def live_stack(api: ApiClient) -> None:
    """会话级探活。栈不可达 → 退出（退出码 4），绝不 skip。"""
    last = "unknown"
    for _ in range(3):
        try:
            resp = api.call("GET", HEALTH_PATH)
            if resp.status_code == 200:
                return
            last = f"HTTP {resp.status_code}"
        except requests.RequestException as exc:
            last = repr(exc)
        time.sleep(2)
    pytest.exit(
        f"E2E 需要活栈：GET {api.base_url}{HEALTH_PATH} 不可达（{last}）。\n"
        f"请先起栈（infrastructure/docker-compose.yml）或设置 E2E_BASE_URL。",
        returncode=4,
    )


@pytest.fixture(scope="session")
def operator_password() -> str:
    pwd = os.environ.get("E2E_OPERATOR_PASSWORD")
    if not pwd:
        pytest.exit("E2E_OPERATOR_PASSWORD 未设置：E2E 无法登录运营账号，不能执行", returncode=4)
    return pwd


ADMIN_USERNAME = "e2e_admin"


@pytest.fixture(scope="session")
def admin_token(api: ApiClient, operator_password: str) -> str:
    """超级管理员 token。

    必须使用 **seed 脚本创建的账号**，而不是 `admin`。

    原因（本轮发现）：库中 operator 账号**没有任何迁移种子**，`DemoDataSeeder`
    也不创建运营账号（只统计），运营账号只能由运行时服务或本 seed 脚本产生。
    此前该 fixture 写的是 `username="admin"` —— 那是某台开发机上手工创建的遗留
    账号，全新库（CI）里根本不存在 → 该 fixture 在 CI 上必然 401，
    **整个 E2E job 从未真正通过过**，只是本机因为有遗留账号而看不出来。
    `e2e_admin` 由 seed 绑定 super_admin 角色，语义等价且可自举。
    """
    data = api.ok(
        "POST",
        "/auth/operator/login",
        json={"username": ADMIN_USERNAME, "password": operator_password},
    )
    token = data.get("accessToken")
    assert token, f"{ADMIN_USERNAME} 登录未返回 accessToken"
    return token


@pytest.fixture
def operator_token(
    api: ApiClient, operator_password: str
) -> Callable[[str], str]:
    """按用户名取运营 token（会话内缓存）。

    可用账号**仅限** tests/e2e/sql/seed_e2e_operators.sql 中创建的 6 个：
    e2e_admin / e2e_platform / e2e_counselor / e2e_station / e2e_matchmaker / e2e_content。
    迁移里**没有任何 operator 种子**（原文"外加迁移 seed 的 admin"是错的，
    已纠正——全新库里没有 admin 这个账号）。用法：`operator_token("e2e_counselor")`。

    传入 seed 之外的用户名等于依赖某台机器上的手工数据，CI 上必然 401；
    `fixture_account_guard_test.py` 会机械拦住这类写法。
    """
    cache: Dict[str, str] = {}

    def _get(username: str) -> str:
        if username not in cache:
            data = api.ok(
                "POST",
                "/auth/operator/login",
                json={"username": username, "password": operator_password},
            )
            token = data.get("accessToken")
            assert token, f"{username} 登录未返回 accessToken"
            cache[username] = token
        return cache[username]

    return _get


@pytest.fixture
def new_user(api: ApiClient) -> Callable[[], Dict[str, Any]]:
    """注册并登录一个全新用户，返回 {'token','refresh_token','user_id','code'}。

    暴露 refresh_token 是为了让刷新链路可被真实栈验证 —— 在 2026-09-22 之前
    没有任何用例覆盖 /auth/refresh，导致「refresh token 用 refresh 密钥签发、
    却用 access 密钥验签」这个必然失败的缺陷长期存活。
    """
    created: list = []

    def _create() -> Dict[str, Any]:
        code = f"e2e-{uuid.uuid4()}"
        data = api.ok("POST", "/auth/wechat/login", json={"code": code})
        user = {
            "token": data.get("accessToken"),
            "refresh_token": data.get("refreshToken"),
            "user_id": data.get("userId"),
            "code": code,
        }
        assert user["token"] and user["user_id"], f"新用户登录返回异常: {data}"
        assert user["refresh_token"], f"登录必须同时返回 refreshToken，否则无法刷新: {data}"
        created.append(user)
        return user

    return _create


@pytest.fixture
def user_token(new_user: Callable[[], Dict[str, Any]]) -> str:
    """独立新用户的 token。"""
    return new_user()["token"]