xst
18890c0437
- Split framework into core/ (config, logger, db, driver factories, screenshot, allure integration) and pages/ (PO layer: base/admin/driver); flatten test files per platform (web/admin, app/driver_android, cross_end) - Slim root conftest.py 656 -> ~100 lines: drop inline CN HTML report, add Allure severity mapping (p0/p1/p2 -> blocker/normal/minor) and environment.properties; support both tests.*-prefixed and bare imports - Preserve TC-D2S-001/002 cross-end E2E cases with business logic unchanged, add allure epic/feature/story/step annotations - Deduplicate: single makereport hook, shared failure-screenshot helpers, single SCREENSHOTS_DIR constant; deterministic auth-state role loading - Move DB credentials from db_helper.py into test_config.yaml (database section) - Move pytest.ini to repo root; switch reporting to allure-pytest (drop pytest-html/junitxml); update requirements.txt accordingly - Delete deprecated scaffolding: test_freight_full_flow + 6 stale page objects, iOS/shipper empty shells, dead utils - Fix flaky vxe-table action-row wait: wait for a visible td instead of hidden placeholder rows (fixed shipper-payment timeouts) - Verified end-to-end against live env: both E2E cases pass (2 passed 3m12s); run artifacts (logs/screenshots/allure) included per repo convention
76 lines
2.3 KiB
Python
76 lines
2.3 KiB
Python
"""MySQL 数据库操作工具 — 测试中的数据库校验/数据修改。
|
||
|
||
连接配置在 tests/config/test_config.yaml 的 database 段维护。
|
||
|
||
Usage:
|
||
from tests.core.db_helper import update_risk_control_status
|
||
|
||
affected = update_risk_control_status("RHJC040026071600021457570057", status=1)
|
||
"""
|
||
|
||
import pymysql
|
||
|
||
from tests.core.config_loader import load_config
|
||
|
||
|
||
def _resolve_config(config: dict | None) -> dict:
|
||
"""显式传入优先，否则读 yaml 的 database 段。"""
|
||
if config is not None:
|
||
return config
|
||
cfg = dict(load_config()["database"])
|
||
cfg.setdefault("charset", "utf8mb4")
|
||
return cfg
|
||
|
||
|
||
def get_connection(config: dict | None = None):
|
||
"""获取数据库连接。"""
|
||
return pymysql.connect(**_resolve_config(config))
|
||
|
||
|
||
def execute_sql(sql: str, params: tuple | None = None, config: dict | None = None) -> int:
|
||
"""执行 SQL（INSERT/UPDATE/DELETE），返回影响行数。"""
|
||
conn = get_connection(config)
|
||
try:
|
||
with conn.cursor() as cursor:
|
||
affected = cursor.execute(sql, params)
|
||
conn.commit()
|
||
return affected
|
||
finally:
|
||
conn.close()
|
||
|
||
|
||
def query_sql(sql: str, params: tuple | None = None, config: dict | None = None) -> list[dict]:
|
||
"""执行查询 SQL，返回结果列表。"""
|
||
conn = get_connection(config)
|
||
try:
|
||
with conn.cursor(pymysql.cursors.DictCursor) as cursor:
|
||
cursor.execute(sql, params)
|
||
return cursor.fetchall()
|
||
finally:
|
||
conn.close()
|
||
|
||
|
||
def update_risk_control_status(order_code: str, status: int = 1, config: dict | None = None) -> int:
|
||
"""更新运单风控状态。
|
||
|
||
执行 SQL:
|
||
UPDATE tms_order_detail_ext SET risk_control_status = <status>
|
||
WHERE order_detail_code IN (
|
||
SELECT `code` FROM `tms_order_detail` WHERE order_code = '<order_code>'
|
||
)
|
||
|
||
Args:
|
||
order_code: 运输单号（对应 tms_order_detail.order_code）
|
||
status: 风控状态值，默认 1
|
||
config: 数据库连接配置（不传则读 yaml database 段）
|
||
Returns:
|
||
影响行数
|
||
"""
|
||
sql = (
|
||
"UPDATE tms_order_detail_ext SET risk_control_status = %s "
|
||
"WHERE order_detail_code IN ("
|
||
" SELECT `code` FROM `tms_order_detail` WHERE order_code = %s"
|
||
")"
|
||
)
|
||
return execute_sql(sql, (status, order_code), config)
|