xst
52b73f456c
- tests/core: 新增 BaseHttpClient 统一 HTTP 封装（直连绕行/超时/日志/自定义编码）； db_helper 追加 Db 长连接类（事务/mogrify/自动重连），既有函数不动 - tests/apis/yun8: 21 个 step 方法按域拆分为 5 个 Mixin（driver/accounting/ audit_payment/receipt/third_party），Yun8Pipeline 24 步编排原样迁移， SQL 全部参数化并集中至 sql_templates - tests/api: 5 阶段有序 P0 E2E 用例（业务失败由 print 改为真实报错）+ README 迁移（删除明文密码段，修正步骤编号漂移） - yun8_sync 目录退役：配置并入 test_config.yaml 的 yun8_api 段； 造数 CLI 保留（python -m tests.apis.yun8.cli，兼容旧 config.yaml 结构） - pytest.ini 注册 api/yun8_sync marker；requirements.txt 补 requests>=2.28 - 静态验证: compileall / 全仓 collect 24 条 / marker 过滤 / CLI --help / 客户端与 pipeline 离线实例化 / SQL 占位符核对 / cross_end 兼容
119 lines
5.5 KiB
Python
119 lines
5.5 KiB
Python
"""统一 HTTP 客户端 — requests.Session 封装。
|
||
|
||
所有 API 客户端（tests/apis/）共用此基类，职责：
|
||
- 单 Session 连接复用 + 请求头管理
|
||
- 直连服务器（禁用系统代理），避免本地代理篡改请求触发服务端 WAF 误判
|
||
- 统一超时 / verify_ssl / 请求响应日志（INFO 摘要、DEBUG 全量）
|
||
- 统一响应处理：非 2xx 抛 HTTPError，成功返回解析后的 JSON dict
|
||
|
||
Usage:
|
||
from tests.core.http_client import BaseHttpClient
|
||
|
||
client = BaseHttpClient("https://api.example.com",
|
||
headers={"deviceid": "xxx"}, timeout=60)
|
||
data = client.get("/v1/resource", params={"id": 1})
|
||
client.set_header("authorization", "Bearer xxx")
|
||
"""
|
||
|
||
import logging
|
||
import urllib.parse
|
||
|
||
import requests
|
||
|
||
# INFO 日志中 path 的最大长度（避免超长 query string 刷屏）
|
||
_PATH_LOG_LIMIT = 200
|
||
|
||
|
||
class BaseHttpClient:
|
||
"""单 Session HTTP 客户端基类。
|
||
|
||
Args:
|
||
base_url: 服务根地址，如 "https://ybxcx.ynyun8.com:8000"
|
||
headers: 初始请求头（跳过空值，与旧 _init_session 行为一致）
|
||
verify_ssl: 是否校验 HTTPS 证书（测试环境自签证书默认关闭）
|
||
timeout: 单请求超时秒数；None 表示不限制（旧版行为）
|
||
name: 日志器后缀，用于区分多套会话（driver/admin/tp）
|
||
"""
|
||
|
||
def __init__(self, base_url: str, headers: dict | None = None,
|
||
verify_ssl: bool = False, timeout: float | None = None,
|
||
name: str = "http"):
|
||
self.base_url = base_url.rstrip("/")
|
||
self.verify_ssl = verify_ssl
|
||
self.timeout = timeout
|
||
self.log = logging.getLogger(f"tests.core.http_client.{name}")
|
||
|
||
self.session = requests.Session()
|
||
for key, value in (headers or {}).items():
|
||
if value:
|
||
self.session.headers[key] = value
|
||
# 禁用系统代理，直连服务器（避免本地代理修改请求导致 WAF 误判 SQL 注入）
|
||
self.session.trust_env = False
|
||
self.session.proxies = {"http": None, "https": None}
|
||
|
||
# ── 请求头管理 ────────────────────────────────────────────────
|
||
def set_header(self, key: str, value: str) -> None:
|
||
"""设置/覆盖一个请求头（允许空值，与旧版无条件覆盖行为一致）。"""
|
||
self.session.headers[key] = value
|
||
|
||
def get_header(self, key: str, default: str = "") -> str:
|
||
"""读取当前会话的一个请求头。"""
|
||
return self.session.headers.get(key, default)
|
||
|
||
# ── 请求方法（返回解析后的 JSON dict）─────────────────────────
|
||
def get(self, path: str, params: dict | None = None) -> dict:
|
||
return self._request("GET", f"{self.base_url}{path}", params=params)
|
||
|
||
def post(self, path: str, params: dict | None = None,
|
||
data: dict | None = None) -> dict:
|
||
return self._request("POST", f"{self.base_url}{path}",
|
||
params=params, data=data)
|
||
|
||
def put(self, path: str, params: dict | None = None,
|
||
data: dict | None = None) -> dict:
|
||
return self._request("PUT", f"{self.base_url}{path}",
|
||
params=params, data=data)
|
||
|
||
def post_json(self, path: str, body: dict,
|
||
params: dict | None = None) -> dict:
|
||
"""POST JSON 请求体（Content-Type: application/json）。"""
|
||
return self._request("POST", f"{self.base_url}{path}",
|
||
params=params, json_body=body)
|
||
|
||
def request_with_encoded_query(self, method: str, path: str,
|
||
params: dict) -> dict:
|
||
"""手工构建 query string 的请求（GET/POST/PUT）。
|
||
|
||
与 get/post/put(params=...) 的区别：query string 由 urlencode 一次性
|
||
编码后拼接为完整 URL 再发给 requests，避免 requests 内部再次编码导致
|
||
URL 值中已有的 %XX（如 AWS Signature 的 %2B）被二次转义为 %25XX，
|
||
触发服务端 WAF 的 SQL 注入误判。
|
||
"""
|
||
qs = urllib.parse.urlencode(params, doseq=True,
|
||
quote_via=urllib.parse.quote)
|
||
full_url = f"{self.base_url}{path}?{qs}"
|
||
self.log.debug("%s 全量 URL: %s", method, full_url)
|
||
return self._request(method, full_url)
|
||
|
||
# ── 内部实现 ─────────────────────────────────────────────────
|
||
def _request(self, method: str, url: str, params: dict | None = None,
|
||
data: dict | None = None, json_body: dict | None = None) -> dict:
|
||
display_path = url[len(self.base_url):][: _PATH_LOG_LIMIT] or "/"
|
||
self.log.info("%s %s", method, display_path)
|
||
resp = self.session.request(
|
||
method, url,
|
||
params=params, data=data, json=json_body,
|
||
verify=self.verify_ssl, timeout=self.timeout,
|
||
)
|
||
resp.raise_for_status()
|
||
payload = resp.json()
|
||
code = payload.get("code") if isinstance(payload, dict) else None
|
||
msg = payload.get("msg", "") if isinstance(payload, dict) else ""
|
||
self.log.info(" ← %s (%.2fs) code=%s msg=%s",
|
||
resp.status_code, resp.elapsed.total_seconds(), code, msg)
|
||
return payload
|
||
|
||
def close(self) -> None:
|
||
"""关闭底层连接池。"""
|
||
self.session.close()
|