WorkBuddy
3c2b967274
CI/CD Pipeline / UI Contracts (source-structure gates) (push) Canceled after 0s
CI/CD Pipeline / E2E (live stack) (push) Canceled after 0s
CI/CD Pipeline / Backend Build & Test (push) Canceled after 0s
CI/CD Pipeline / Admin Web Build (push) Canceled after 0s
CI/CD Pipeline / Mini Program Compile Check (push) Canceled after 0s
CI/CD Pipeline / Migration Validation (push) Canceled after 0s
CI/CD Pipeline / Test Shell Guard (Phase 23 Gate) (push) Canceled after 0s
217 lines
8.1 KiB
Python
217 lines
8.1 KiB
Python
"""
|
||
善缘丹阳 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:
|
||
# 登录接口会设置运营 HttpOnly Cookie；无 token 用例必须真正模拟未认证请求。
|
||
if token is None:
|
||
self.session.cookies.clear()
|
||
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"
|
||
# 本机活栈可通过环境变量切换到已授权的演示运营账号；CI 默认仍使用 seed 账号。
|
||
LOGIN_USERNAME = os.environ.get("E2E_OPERATOR_USERNAME", ADMIN_USERNAME)
|
||
|
||
|
||
@pytest.fixture(scope="session")
|
||
def admin_password(operator_password: str) -> str:
|
||
return os.environ.get("E2E_ADMIN_PASSWORD", operator_password)
|
||
|
||
|
||
@pytest.fixture(scope="session")
|
||
def admin_token(api: ApiClient, operator_password: str, admin_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": LOGIN_USERNAME, "password": admin_password},
|
||
)
|
||
token = data.get("accessToken")
|
||
assert token, f"{LOGIN_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"]
|