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,
    ) -> requests.Response:
        headers = {"Authorization": f"Bearer {token}"} if token else {}
        return self.session.request(
            method,
            f"{self.base_url}{path}",
            headers=headers,
            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


@pytest.fixture(scope="session")
def admin_token(api: ApiClient, operator_password: str) -> str:
    data = api.ok(
        "POST",
        "/auth/operator/login",
        json={"username": "admin", "password": operator_password},
    )
    token = data.get("accessToken")
    assert token, "admin 登录未返回 accessToken"
    return token


@pytest.fixture
def operator_token(
    api: ApiClient, operator_password: str
) -> Callable[[str], str]:
    """按用户名取运营 token（会话内缓存）。用法：`operator_token("counselor01")`。"""
    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','user_id','code'}。"""
    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"),
            "user_id": data.get("userId"),
            "code": code,
        }
        assert user["token"] and user["user_id"], f"新用户登录返回异常: {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"]