From f75a6f5ec9a4bfcce309a13f5512b5a5b0a26f01 Mon Sep 17 00:00:00 2001 From: xst Date: Wed, 22 Jul 2026 17:18:52 +0800 Subject: [PATCH] feat(cross_end): add driver-to-settlement E2E test with SQL retry - Add DriverFreightFlowPage (Android): 28-step flow covering accept/load/unload - Add FreightAccountingPage (Web): accounting/audit/finance-payment/shipper-payment - Add db_helper: update_risk_control_status via pymysql - Fix ensure_logged_in: coordinate tap on 'mine' tab + login check - Add test_driver_to_settlement.py: full cross-end E2E with SQL retry Co-authored-by: xst --- requirements.txt | 1 + tests/app/_shared.py | 92 ++++-- .../pages/admin/freight_accounting_page.py | 279 ++++++++++++++++++ .../pages/app/driver_freight_flow_page.py | 212 +++++++++++++ .../tests/test_driver_to_settlement.py | 129 ++++++++ tests/utils/db_helper.py | 72 +++++ 6 files changed, 763 insertions(+), 22 deletions(-) create mode 100644 tests/cross_end/pages/admin/freight_accounting_page.py create mode 100644 tests/cross_end/pages/app/driver_freight_flow_page.py create mode 100644 tests/cross_end/tests/test_driver_to_settlement.py create mode 100644 tests/utils/db_helper.py diff --git a/requirements.txt b/requirements.txt index 9cb6919..7ea8276 100644 --- a/requirements.txt +++ b/requirements.txt @@ -5,3 +5,4 @@ pyantiword playwright>=1.40 Pillow>=10.0 # 截图像素对比 (result-reporter Agent) # Appium-Python-Client>=3.0 # 取消注释以启用移动端自动化 +pymysql>=1.0 # 数据库操作（测试中更新风控状态等） diff --git a/tests/app/_shared.py b/tests/app/_shared.py index 03e7762..2adb763 100644 --- a/tests/app/_shared.py +++ b/tests/app/_shared.py @@ -54,11 +54,14 @@ def make_ensure_logged_in( ): """创建一个 ensure_logged_in 闭包。 + 策略：坐标点击底部"我的"tab（第4个图标）→ 检查"登录/注册"文本。 + 无法通过 resourceId 或 text 定位底部导航时用坐标点击兜底。 + Args: section_config: App 端配置 config: 全局配置 account_role: 默认账号角色名（如 "driver"、"shipper"） - home_indicator: 首页元素的 accessibility_id + home_indicator: 未使用，保留兼容 """ from selenium.webdriver.support.ui import WebDriverWait from selenium.webdriver.support import expected_conditions as EC @@ -68,42 +71,87 @@ def make_ensure_logged_in( skip_login = section_config.get("skip_login", True) wait = WebDriverWait(driver, timeout=timeout_s) - try: - wait.until( - EC.visibility_of_element_located((AppiumBy.ACCESSIBILITY_ID, home_indicator)) - ) - return - except Exception: - pass + driver.implicitly_wait(3) + # 坐标点击底部"我的"tab（4 tab 均分宽度，第4个中心 x≈0.875*w, y≈h-50） try: - login_input = driver.find_element(AppiumBy.ACCESSIBILITY_ID, "login_phone_input") - if login_input.is_displayed(): + w = driver.get_window_size()["width"] + h = driver.get_window_size()["height"] + driver.tap([(int(w * 0.875), h - 50)]) + driver.implicitly_wait(1) + except Exception as e: + raise RuntimeError(f"无法点击底部导航栏: {e}") + + # 判断登录态：检查"登录/注册"文本 + login_check = (AppiumBy.ANDROID_UIAUTOMATOR, 'new UiSelector().text("登录/注册")') + try: + el = driver.find_element(*login_check) + if el.is_displayed(): if skip_login: raise RuntimeError( - "APP 在登录页，未检测到登录态！请在设备上手动完成一次登录（含验证码），" + "APP 未登录！请在设备上手动完成一次登录（含验证码），" "之后重新运行用例即可。\n如需自动化登录，请设置 skip_login: false。" ) - account = config["accounts"].get(account_role, {}) - login_input.send_keys(account.get("phone", "")) - driver.find_element( - AppiumBy.ACCESSIBILITY_ID, "login_password_input" - ).send_keys(account.get("password", "")) - driver.find_element(AppiumBy.ACCESSIBILITY_ID, "login_submit_btn").click() - wait.until( - EC.visibility_of_element_located((AppiumBy.ACCESSIBILITY_ID, home_indicator)) - ) + _perform_ui_login(driver, config["accounts"].get(account_role, {}), wait) return except RuntimeError: raise + except Exception: + pass # 没找到"登录/注册" → 已登录 + + # 回到首页 + home_tab = ( + AppiumBy.ANDROID_UIAUTOMATOR, + 'new UiSelector().resourceId("com.arpa.ynchenggangdriver:id/fixed_bottom_navigation_icon").instance(0)', + ) + try: + wait.until(EC.visibility_of_element_located(home_tab)) + driver.find_element(*home_tab).click() except Exception: pass - raise RuntimeError("无法确定 APP 的登录状态！请确保 APP 已启动且首页或登录页可见。") - return _ensure_logged_in +def _perform_ui_login(driver, account: dict, wait) -> None: + """执行 UI 自动登录（skip_login: false 时使用）。""" + from appium.webdriver.common.appiumby import AppiumBy + + phone = account.get("phone", "") + password = account.get("password", "") + + # 点击登录入口 + login_entry = (AppiumBy.ANDROID_UIAUTOMATOR, 'new UiSelector().text("登录/注册")') + wait.until(lambda d: d.find_element(*login_entry)) + driver.find_element(*login_entry).click() + driver.implicitly_wait(1) + + # 填写手机号 + phone_input = (AppiumBy.ID, "com.arpa.ynchenggangdriver:id/ed_phone") + wait.until(lambda d: d.find_element(*phone_input)) + driver.find_element(*phone_input).send_keys(phone) + + # 填写密码 + password_input = (AppiumBy.ID, "com.arpa.ynchenggangdriver:id/ed_pwd") + driver.find_element(*password_input).send_keys(password) + + # 点击登录 + login_btn = (AppiumBy.ID, "com.arpa.ynchenggangdriver:id/tv_login") + driver.find_element(*login_btn).click() + driver.implicitly_wait(2) + + # 回到首页 + home_tab = ( + AppiumBy.ANDROID_UIAUTOMATOR, + 'new UiSelector().resourceId("com.arpa.ynchenggangdriver:id/fixed_bottom_navigation_icon").instance(0)', + ) + try: + wait.until(lambda d: d.find_element(*home_tab)) + driver.find_element(*home_tab).click() + except Exception: + pass + + def save_failure_screenshot(driver, section_config, platform, request): """失败自动截图（各端 conftest teardown 中调用）。""" if ( diff --git a/tests/cross_end/pages/admin/freight_accounting_page.py b/tests/cross_end/pages/admin/freight_accounting_page.py new file mode 100644 index 0000000..51aee74 --- /dev/null +++ b/tests/cross_end/pages/admin/freight_accounting_page.py @@ -0,0 +1,279 @@ +"""管理端 Page Object — 运费核算 + 运单审核 + 垫资打款 完整流程。 + +基于 resources/selectors/正常流程-web端.md 中的元素定位。 +使用直接 hash URL 导航（比菜单点击更可靠），按钮选择器带 fallback 兜底。 + +URL 路由: + 运费核算: #/tms/Carrier/freight + 运单审核: #/tms/Carrier/OrderDetailAudit + 垫资-财务打款: #/tms/Carrier/dzFinancialapproval + 垫资-货主打款: #/tms/Carrier/dzApplicationForPayment +""" + +from playwright.sync_api import Page + +from tests.web.admin.pages.base_page import BasePage + + +class FreightAccountingPage(BasePage): + """管理端 - 运费核算 / 运单审核 / 垫资打款。""" + + def __init__(self, page: Page, admin_base_url: str = ""): + super().__init__(page) + self.admin_base_url = admin_base_url + + # ════════════════════════════════════════════════════════════ + # URL 导航 + # ════════════════════════════════════════════════════════════ + + def _go(self, hash_route: str): + """直接 hash URL 跳转，比菜单点击稳定。""" + url = f"{self.admin_base_url}/#/{hash_route.lstrip('#').lstrip('/')}" + self.page.goto(url, wait_until="domcontentloaded") + self.page.wait_for_timeout(2000) + + # ════════════════════════════════════════════════════════════ + # 通用操作 + # ════════════════════════════════════════════════════════════ + + def _fill_order_no(self, order_no: str): + """搜索栏输入运输单号（多个同名 input 时取第一个）。""" + input_el = self.page.locator('input[placeholder="请输入运输单号"]').first + self.wait_visible(input_el) + input_el.clear() + input_el.fill(order_no) + + def _click_search(self): + """点击查询按钮。""" + btn = self.page.locator("button:has-text('查询')").first + self.wait_visible(btn) + btn.click() + self.page.wait_for_timeout(1000) + + def _table_has_data(self) -> bool: + """检查查询结果是否有数据。""" + try: + self.page.wait_for_selector("table tbody tr", timeout=3_000) + return True + except Exception: + return False + + def _click_table_action(self, button_index: int): + """点击表格固定列操作按钮（纯图标 按钮，selector 文档路径 + fallback）。 + + vxe-table 操作列按钮结构: td > div > button:nth-child(N) > span > i + """ + # 精确路径（selector 文档原始定位） + sel = ( + ".vxe-table--fixed-right-wrapper.scrolling--middle " + ".vxe-table--body-wrapper.fixed-right--wrapper " + "table > tbody > tr:nth-child(1) > td > div " + f"> button:nth-child({button_index}) > span > i" + ) + # 宽松路径（当 DOM 结构略有变化时兜底） + sel_fallback = ( + ".vxe-table--fixed-right-wrapper " + "table > tbody > tr:nth-child(1) " + f"> td:last-child > div > button:nth-child({button_index})" + ) + + for attempt in range(3): + try: + self.page.wait_for_selector("table tbody tr", timeout=5_000) + self.page.wait_for_timeout(300) + btn = self.page.locator(sel).first + if btn.count() == 0: + btn = self.page.locator(sel_fallback).first + self.wait_visible(btn, timeout_ms=5_000) + btn.click() + self.page.wait_for_timeout(800) + return + except Exception: + if attempt == 2: + raise + self.page.wait_for_timeout(1_000) + + def _modal(self): + """返回当前可见弹窗内容容器（.first 避免 strict mode）。""" + return self.page.locator( + ".vxe-modal--wrapper.is--visible .vxe-modal--body .vxe-modal--content" + ).first + + def _scroll_modal(self): + """滚动弹窗到底部。""" + modal = self._modal() + self.wait_visible(modal) + modal.evaluate("el => el.scrollTop = el.scrollHeight") + self.page.wait_for_timeout(500) + + def _retry_click_modal_btn(self, locator, timeout_ms: int = 5_000, + retries: int = 3, wait_after: int = 1500): + """弹窗内按钮带重试点击（每次重试重新获取元素，防 DOM 置换）。""" + for attempt in range(retries): + el = locator() if callable(locator) else locator + try: + self.wait_visible(el, timeout_ms=timeout_ms) + el.click() + self.page.wait_for_timeout(wait_after) + return + except Exception: + if attempt == retries - 1: + raise + self.page.wait_for_timeout(1_000) + + # ════════════════════════════════════════════════════════════ + # 运费核算 + # ════════════════════════════════════════════════════════════ + + def do_freight_accounting(self, order_no: str, retry_sql=None): + """运费核算：导航→搜索→（无数据时 SQL 重试）→编辑→核算。 + + Args: + order_no: 运输单号 + retry_sql: 可调用对象，首次查询无数据时调用 + """ + self._go("tms/Carrier/freight") + self._fill_order_no(order_no) + self._click_search() + + # 无数据 → 执行 SQL 重试 + if not self._table_has_data(): + if retry_sql: + retry_sql() + self._fill_order_no(order_no) + self._click_search() + + if not self._table_has_data(): + raise RuntimeError(f"运费核算查询无数据！单号: {order_no}") + + # 点击编辑 (button_index=4, 来自 selector 文档 Step 3) + self._click_table_action(button_index=4) + + # 核算按钮 (selector 文档 Step 4) + btn = self._modal().locator( + "form > div:nth-child(4) button:nth-child(4) span" + ).first + self._retry_click_modal_btn(btn, wait_after=1500) + self.screenshot("freight_accounting_done") + + # ════════════════════════════════════════════════════════════ + # 运单审核 + # ════════════════════════════════════════════════════════════ + + def do_waybill_audit(self, order_no: str): + """运单审核：导航→搜索→审核→滚动弹窗→审核通过。""" + self._go("tms/Carrier/OrderDetailAudit") + self._fill_order_no(order_no) + self._click_search() + + # 点击审核 (button_index=2, selector 文档 Step 10) + self._click_table_action(button_index=2) + + # 滚动弹窗 (selector 文档 Step 11) + self._scroll_modal() + + # 审核通过 (selector 文档 Step 12) + btn = self._modal().locator( + "._btns > button.el-button.el-button--primary.el-button--medium > span" + ).first + self._retry_click_modal_btn(btn, wait_after=1500) + self.screenshot("waybill_audit_done") + + # ════════════════════════════════════════════════════════════ + # 垫资-财务打款 + # ════════════════════════════════════════════════════════════ + + def do_finance_payment(self, order_no: str, verify_code: str = "1234"): + """垫资-财务打款：导航→搜索→在线支付→滚动→勾选光大/手机→验证码→确认。""" + self._go("tms/Carrier/dzFinancialapproval") + self._fill_order_no(order_no) + self._click_search() + + # 在线支付 (button_index=2, selector 文档 Step 18) + self._click_table_action(button_index=2) + self._scroll_modal() + + # 勾选光大银行 (selector 文档 Step 20) + try: + radio = self._modal().locator( + "form > div:nth-child(9) > div > div:nth-child(2) " + "> label > span.el-radio__input > span" + ).first + self._retry_click_modal_btn(radio, timeout_ms=3_000, wait_after=300) + except Exception: + pass + + # 勾选手机号 (selector 文档 Step 21) + try: + radio = self.page.locator( + "[id^='el-collapse-content-'] > div > div:nth-child(3) " + "> label > span.el-radio__input > span" + ).first + self._retry_click_modal_btn(radio, timeout_ms=3_000, wait_after=300) + except Exception: + pass + + # 获取验证码 (selector 文档 Step 22) + try: + btn = self._modal().locator( + "form > div.el-row > div.el-col.el-col-4 button span" + ).first + self._retry_click_modal_btn(btn, timeout_ms=3_000, wait_after=500) + except Exception: + pass + + # 输入验证码 (selector 文档 Step 23) + try: + inp = self._modal().locator( + "form > div.el-row > div.el-col.el-col-20 input" + ).first + self.wait_visible(inp) + inp.clear() + inp.fill(verify_code) + except Exception: + pass + + # 确认打款 (selector 文档 Step 24) + btn = self._modal().locator( + "form > div:nth-child(12) > div > button span" + ).first + self._retry_click_modal_btn(btn, wait_after=1500) + self.screenshot("finance_payment_done") + + # ════════════════════════════════════════════════════════════ + # 垫资-货主打款 + # ════════════════════════════════════════════════════════════ + + def do_shipper_payment(self, order_no: str): + """垫资-货主打款：导航→搜索→打款→勾选光大→确认打款→确认弹窗。""" + self._go("tms/Carrier/dzApplicationForPayment") + self._fill_order_no(order_no) + self._click_search() + + # 打款 (button_index=2, selector 文档 Step 30) + self._click_table_action(button_index=2) + + # 勾选光大 (selector 文档 Step 31) + try: + radio = self._modal().locator( + "form > div > div:nth-child(3) > div > div > div:nth-child(3) " + "> label > span.el-radio__label" + ).first + self._retry_click_modal_btn(radio, timeout_ms=3_000, wait_after=300) + except Exception: + pass + + # 确认打款 (selector 文档 Step 31-2) + btn = self._modal().locator("> div > div > button > span").first + self._retry_click_modal_btn(btn, wait_after=1000) + + # message-box 确认 (selector 文档 Step 32) + try: + msg = self.page.locator( + "body > div.el-message-box__wrapper .el-message-box__btns " + "> button.el-button.el-button--default.el-button--small.el-button--primary > span" + ) + self._retry_click_modal_btn(msg, timeout_ms=5_000, wait_after=1500) + except Exception: + pass + self.screenshot("shipper_payment_done") diff --git a/tests/cross_end/pages/app/driver_freight_flow_page.py b/tests/cross_end/pages/app/driver_freight_flow_page.py new file mode 100644 index 0000000..4f38522 --- /dev/null +++ b/tests/cross_end/pages/app/driver_freight_flow_page.py @@ -0,0 +1,212 @@ +"""司机 APP 货运流程 Page Object — 货源→接单→装货→卸货→获取单号。 + +基于 resources/selectors/正常流程-司机.md 中的元素定位。 +使用 Appium UiAutomator2 定位策略。 +""" + +from appium.webdriver.common.appiumby import AppiumBy +from selenium.webdriver.support.ui import WebDriverWait +from selenium.webdriver.support import expected_conditions as EC + +from tests.app._base_page import AppBasePage + + +class DriverFreightFlowPage(AppBasePage): + """司机端 APP - 货运完整流程（货源→接单→装货→卸货→获取单号）。""" + + def __init__(self, driver, platform: str = "android"): + super().__init__(driver, platform) + + # ── 底部导航 ── + # 0=首页, 1=货源, 2=消息, 3=我的 + self.freight_source_tab = ( + AppiumBy.ANDROID_UIAUTOMATOR, + 'new UiSelector().resourceId("com.arpa.ynchenggangdriver:id/fixed_bottom_navigation_icon").instance(1)', + ) + + # ── 货源单列表 ── + self.freight_order_item = ( + AppiumBy.ANDROID_UIAUTOMATOR, + 'new UiSelector().className("android.widget.RelativeLayout").instance(3)', + ) + + # ── 数量 ── + self.quantity_input = (AppiumBy.ID, "com.arpa.ynchenggangdriver:id/et_num") + + # ── 协议勾选 ── + self.agreement_checkbox = (AppiumBy.ID, "com.arpa.ynchenggangdriver:id/check_xieyi") + + # ── 接单 ── + self.accept_order_btn = (AppiumBy.ID, "com.arpa.ynchenggangdriver:id/tv_bian") + + # ── 系统弹窗取消 ── + self.cancel_btn = (AppiumBy.ID, "android:id/button2") + + # ── 吨位 ── + self.weight_input = (AppiumBy.ID, "com.arpa.ynchenggangdriver:id/et_dun") + + # ── 上传图片 ── + self.select_image_btn = ( + AppiumBy.ANDROID_UIAUTOMATOR, + 'new UiSelector().className("android.widget.ImageView").instance(2)', + ) + self.album_btn = (AppiumBy.ANDROID_UIAUTOMATOR, 'new UiSelector().text("相册")') + self.image_checkbox = (AppiumBy.ID, "com.arpa.ynchenggangdriver:id/cb_check") + self.done_btn = (AppiumBy.ID, "com.arpa.ynchenggangdriver:id/btn_ok") + + # ── 装货/卸货确认 ── + self.load_unload_confirm = (AppiumBy.ID, "com.arpa.ynchenggangdriver:id/tv_nick") + + # ── 合同 ── + self.contract_webview = (AppiumBy.ID, "com.arpa.ynchenggangdriver:id/textHtml") + self.acknowledge_btn = (AppiumBy.ID, "com.arpa.ynchenggangdriver:id/tv_affirm") + self.confirm_btn = (AppiumBy.ID, "com.arpa.ynchenggangdriver:id/ok_tv3") + + # ════════════════════════════════════════════════════════════ + # 基础操作封装 + # ════════════════════════════════════════════════════════════ + + def _wait_and_click(self, locator, timeout_s: float = 10.0): + wait = WebDriverWait(self.driver, timeout=timeout_s) + wait.until(EC.visibility_of_element_located(locator)) + self.driver.find_element(*locator).click() + + def _wait_and_fill(self, locator, value: str, timeout_s: float = 10.0): + wait = WebDriverWait(self.driver, timeout=timeout_s) + wait.until(EC.visibility_of_element_located(locator)) + el = self.driver.find_element(*locator) + el.clear() + el.send_keys(value) + + def _tap_by_text(self, text: str, timeout_s: float = 10.0): + locator = (AppiumBy.ANDROID_UIAUTOMATOR, f'new UiSelector().text("{text}")') + self._wait_and_click(locator, timeout_s) + + def _safe_tap_by_text(self, text: str, timeout_s: float = 5.0): + """容错点击：元素不存在时静默跳过。""" + try: + self._tap_by_text(text, timeout_s) + except Exception: + pass + + def _swipe_up(self, locator=None, duration_ms: int = 800): + """向上滑动。指定 locator 时在元素内滑动，否则全屏滑动。""" + if locator: + try: + wait = WebDriverWait(self.driver, timeout=5.0) + el = wait.until(EC.presence_of_element_located(locator)) + size = el.size + loc = el.location + start_x = loc["x"] + size["width"] // 2 + start_y = loc["y"] + int(size["height"] * 0.8) + end_y = loc["y"] + int(size["height"] * 0.2) + self.driver.swipe(start_x, start_y, start_x, end_y, duration_ms) + return + except Exception: + pass + # 全屏滑动 + size = self.driver.get_window_size() + self.driver.swipe( + start_x=size["width"] // 2, + start_y=int(size["height"] * 0.7), + end_x=size["width"] // 2, + end_y=int(size["height"] * 0.3), + duration=duration_ms, + ) + + # ════════════════════════════════════════════════════════════ + # 接单流程 (步骤 1-5) + # ════════════════════════════════════════════════════════════ + + def accept_order_flow(self, quantity: str = "1"): + """接单完整流程：货源 tab → 货源单 → 数量 → 协议 → 接单。""" + self._wait_and_click(self.freight_source_tab) # ① 点击货源 + self.driver.implicitly_wait(1) + self._wait_and_click(self.freight_order_item) # ② 点击货源单 + self.driver.implicitly_wait(2) + self._wait_and_fill(self.quantity_input, quantity) # ③ 填写数量 + self._wait_and_click(self.agreement_checkbox) # ④ 勾选协议 + self._wait_and_click(self.accept_order_btn) # ⑤ 接单 + self.driver.implicitly_wait(2) + + # ════════════════════════════════════════════════════════════ + # 装货流程 (步骤 6-17) + # ════════════════════════════════════════════════════════════ + + def load_goods_flow(self, weight: str = "10"): + """装货完整流程。""" + self._tap_by_text("装货") # ⑥ 点击装货 + self.driver.implicitly_wait(1) + self._wait_and_click(self.cancel_btn) # ⑦ 点击取消（系统弹窗） + self.driver.implicitly_wait(1) + self._wait_and_fill(self.weight_input, weight) # ⑧ 输入吨位 + self._wait_and_click(self.select_image_btn) # ⑨ 选择图片 + self.driver.implicitly_wait(1) + self._wait_and_click(self.album_btn) # ⑩ 相册 + self.driver.implicitly_wait(1) + self._wait_and_click(self.image_checkbox) # ⑪ 勾选图片 + self._wait_and_click(self.done_btn) # ⑫ 完成 + self.driver.implicitly_wait(2) + self._wait_and_click(self.agreement_checkbox) # ⑬ 勾选协议 + self._wait_and_click(self.load_unload_confirm) # ⑭ 装货确认 + self.driver.implicitly_wait(1) + self._swipe_up(self.contract_webview) # ⑮ 滑动合同 + self.driver.implicitly_wait(1) + self._safe_tap_by_text("已知晓", timeout_s=5.0) # ⑯ 已知晓（容错） + self._safe_tap_by_text("确定", timeout_s=5.0) # ⑰ 确定（容错） + self.driver.implicitly_wait(2) + + # ════════════════════════════════════════════════════════════ + # 卸货流程 (步骤 19-27) + # ════════════════════════════════════════════════════════════ + + def unload_goods_flow(self, weight: str = "10"): + """卸货完整流程。""" + self._tap_by_text("卸货") # ⑲ 点击卸货 + self.driver.implicitly_wait(1) + self._wait_and_click(self.cancel_btn) # ⑳ 点击取消（系统弹窗） + self.driver.implicitly_wait(1) + self._wait_and_fill(self.weight_input, weight) # ㉑ 输入吨位 + self._wait_and_click(self.select_image_btn) # ㉒ 选择图片 + self.driver.implicitly_wait(1) + self._wait_and_click(self.album_btn) # ㉓ 相册 + self.driver.implicitly_wait(1) + self._wait_and_click(self.image_checkbox) # ㉔ 勾选图片 + self._wait_and_click(self.done_btn) # ㉕ 完成 + self.driver.implicitly_wait(2) + self._wait_and_click(self.load_unload_confirm) # ㉖ 卸货确认 + self.driver.implicitly_wait(1) + self._safe_tap_by_text("确定", timeout_s=5.0) # ㉗ 确定（容错） + self.driver.implicitly_wait(2) + + # ════════════════════════════════════════════════════════════ + # 获取单号 (步骤 28) + # ════════════════════════════════════════════════════════════ + + def get_transport_order_no(self, timeout_s: float = 10.0) -> str: + """获取运输单号。从"运输单号：RHJC..."中正则提取纯单号。""" + import re + + locator = ( + AppiumBy.ANDROID_UIAUTOMATOR, + 'new UiSelector().textContains("运输单号")', + ) + wait = WebDriverWait(self.driver, timeout=timeout_s) + el = wait.until(EC.visibility_of_element_located(locator)) + full_text = el.text + match = re.search(r"RHJC\S+", full_text) + if match: + return match.group() + return full_text.replace("运输单号：", "").replace("运输单号:", "").strip() + + # ════════════════════════════════════════════════════════════ + # 全流程 + # ════════════════════════════════════════════════════════════ + + def full_driver_flow(self, quantity: str = "1", load_weight: str = "10", + unload_weight: str = "10") -> str: + """执行完整司机端流程：接单→装货→卸货→返回运输单号。""" + self.accept_order_flow(quantity) + self.load_goods_flow(load_weight) + self.unload_goods_flow(unload_weight) + return self.get_transport_order_no() diff --git a/tests/cross_end/tests/test_driver_to_settlement.py b/tests/cross_end/tests/test_driver_to_settlement.py new file mode 100644 index 0000000..c6eb738 --- /dev/null +++ b/tests/cross_end/tests/test_driver_to_settlement.py @@ -0,0 +1,129 @@ +"""跨端 E2E 测试 — 司机接单装货卸货 + 管理端核算打款完整流程。 + +业务流程: + 司机 APP (Android): + 1. 接单: 点击货源 → 选择货源单 → 填写数量 → 勾选协议 → 接单 + 2. 装货: 装货 → 取消弹窗 → 吨位 → 选图 → 相册 → 勾图 → 完成 → 勾选 → 装货确认 → 合同 → 已知晓 → 确定 + 3. 卸货: 卸货 → 取消弹窗 → 吨位 → 选图 → 相册 → 勾图 → 完成 → 卸货确认 → 确定 + 4. 获取运输单号 + + Web 管理端: + 5. 运费核算: URL 跳转 → 输入单号 → 查询 →（无数据时 SQL 重试）→ 编辑 → 核算 + 6. 运单审核: URL 跳转 → 输入单号 → 查询 → 审核 → 滚动弹窗 → 审核通过 + 7. 垫资-财务打款: URL 跳转 → 输入单号 → 查询 → 在线支付 → 滚动 → + 勾选光大 → 勾选手机号 → 获取验证码 → 输入验证码 → 确认打款 + 8. 垫资-货主打款: URL 跳转 → 输入单号 → 查询 → 打款 → 勾选光大 → 确认打款 → 确认弹窗 + +运行: + pytest tests/cross_end/tests/test_driver_to_settlement.py -m cross_end -v -s + +前置条件: + - Appium server: http://localhost:4723 + - Android 模拟器已启动，司机 APP 已安装并登录过一次 (noReset) + - 管理端 auth 态: python tests/web/admin/save_auth.py --role super_admin + - MySQL 数据库可连接 (pymysql) +""" + +import pytest + +from tests.cross_end.pages.admin.freight_accounting_page import FreightAccountingPage +from tests.cross_end.pages.app.driver_freight_flow_page import DriverFreightFlowPage +from tests.utils.db_helper import update_risk_control_status + +pytestmark = [pytest.mark.cross_end, pytest.mark.p0, pytest.mark.smoke] + + +class TestDriverToSettlement: + """完整业务流程: 司机 APP 接单→装货→卸货 → Web 管理端核算→打款。 + + 测试覆盖: + - 司机 APP: 货源浏览、接单、装货（含拍照上传、合同确认）、卸货（含拍照上传） + - Web 管理端: 运费核算、运单审核、垫资-财务打款（含验证码）、垫资-货主打款 + - 数据库: SQL 更新风控状态（运费核算查询无数据时自动执行） + """ + + def test_driver_to_settlement_full_flow( + self, + admin_page, + admin_config, + app_driver, + ensure_admin_login, + ensure_driver_login, + cross_end_context, + logger, + ): + """验证从司机 APP 接单到管理端核算打款的完整跨端业务流程。""" + logger.info("=" * 60) + logger.info("跨端 E2E: 司机APP → 管理端核算打款 完整流程开始") + logger.info("=" * 60) + + # ── 登录确认 ── + ensure_admin_login("super_admin") + ensure_driver_login(app_driver) + + # ── 初始化 Page Objects ── + driver_flow = DriverFreightFlowPage(app_driver) + admin_accounting = FreightAccountingPage(admin_page, admin_base_url=admin_config["base_url"]) + + # ════════════════════════════════════════════════════════════ + # Phase 1: 司机 Android APP — 接单 + 装货 + 卸货 + # ════════════════════════════════════════════════════════════ + + logger.info("Phase 1: 司机 APP — 开始接单流程") + driver_flow.accept_order_flow(quantity="1") + driver_flow.screenshot("step_accept_order") + logger.info(" -> 接单流程完成") + + logger.info("Phase 1: 司机 APP — 开始装货流程") + driver_flow.load_goods_flow(weight="10") + driver_flow.screenshot("step_load_goods") + logger.info(" -> 装货流程完成") + + logger.info("Phase 1: 司机 APP — 开始卸货流程") + driver_flow.unload_goods_flow(weight="10") + driver_flow.screenshot("step_unload_goods") + logger.info(" -> 卸货流程完成") + + logger.info("Phase 1: 司机 APP — 获取运输单号") + transport_order_no = driver_flow.get_transport_order_no() + cross_end_context["order_no"] = transport_order_no + logger.info(f" -> 运输单号: {transport_order_no}") + + # ════════════════════════════════════════════════════════════ + # Phase 2: Web 管理端 — 核算 + 审核 + 打款 + # ════════════════════════════════════════════════════════════ + + logger.info("Phase 2: Web管理端 — 运费核算（首次查询无数据时自动 SQL 重试）") + admin_accounting.do_freight_accounting( + transport_order_no, + retry_sql=lambda: update_risk_control_status(transport_order_no, status=1), + ) + cross_end_context["accounting_done"] = True + logger.info(f" -> 运费核算完成: {transport_order_no}") + + logger.info("Phase 2: Web管理端 — 运单审核") + admin_accounting.do_waybill_audit(transport_order_no) + cross_end_context["review_done"] = True + logger.info(f" -> 运单审核完成: {transport_order_no}") + + logger.info("Phase 2: Web管理端 — 垫资-财务打款") + admin_accounting.do_finance_payment(transport_order_no, verify_code="1234") + logger.info(f" -> 垫资-财务打款完成: {transport_order_no}") + + logger.info("Phase 2: Web管理端 — 垫资-货主打款") + admin_accounting.do_shipper_payment(transport_order_no) + logger.info(f" -> 垫资-货主打款完成: {transport_order_no}") + + # ════════════════════════════════════════════════════════════ + # 最终验证 + # ════════════════════════════════════════════════════════════ + logger.info("=" * 60) + logger.info("跨端 E2E: 司机APP → 管理端核算打款 完整流程结束!") + logger.info(f" 运输单号: {cross_end_context['order_no']}") + logger.info(f" 核算完成: {cross_end_context['accounting_done']}") + logger.info(f" 审核完成: {cross_end_context['review_done']}") + logger.info("=" * 60) + + assert cross_end_context["order_no"], "运输单号不能为空" + assert cross_end_context["accounting_done"], "核算未完成" + assert cross_end_context["review_done"], "审核未完成" diff --git a/tests/utils/db_helper.py b/tests/utils/db_helper.py new file mode 100644 index 0000000..110d633 --- /dev/null +++ b/tests/utils/db_helper.py @@ -0,0 +1,72 @@ +"""MySQL 数据库操作工具 — 测试中的数据库校验/数据修改。 + +Usage: + from tests.utils.db_helper import update_risk_control_status + + affected = update_risk_control_status("RHJC040026071600021457570057", status=1) +""" + +import pymysql + +DEFAULT_DB_CONFIG = { + "host": "36.137.133.24", + "port": 3306, + "user": "ntocc", + "password": "Rgxi3!kD9jjXpgzx", + "database": "ntocc", + "charset": "utf8mb4", +} + + +def get_connection(config: dict | None = None): + """获取数据库连接。""" + cfg = config or DEFAULT_DB_CONFIG + return pymysql.connect(**cfg) + + +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 = + WHERE order_detail_code IN ( + SELECT `code` FROM `tms_order_detail` WHERE order_code = '' + ) + + Args: + order_code: 运输单号（对应 tms_order_detail.order_code） + status: 风控状态值，默认 1 + config: 数据库连接配置 + 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)