feat: complete CONTRACT_CREATE page flow with Selenium adapter
This commit is contained in:
34BFF76918A9449ABAB323149DDCEA73
parent
ca79659e3e
commit
5824279fd5
|
|
@ -21,6 +21,12 @@ state_center_web/dist/
|
|||
|
||||
# Local runtime / agent
|
||||
.runtime/
|
||||
rpa_yisaiqi/screenshots/
|
||||
rpa-chrome-profile/
|
||||
**/chrome-profile/
|
||||
.tmp/
|
||||
.grok/
|
||||
.grokignore
|
||||
.claude/
|
||||
.superpowers/
|
||||
|
||||
|
|
|
|||
|
|
@ -57,10 +57,20 @@ python -m rpa_yisaiqi.contract_create_node `
|
|||
|
||||
## 真实合同起草运行
|
||||
|
||||
先准备合同起草输入 JSON，参考：
|
||||
客商必须用合同测试系统里**真实存在**的操作码。`CUST-YYC-001` 这类占位符会在选客商时搜不到，入参校验会直接拒绝。
|
||||
|
||||
- `examples/contract_draft_yyc.json`
|
||||
- `examples/contract_draft_xyy.json`
|
||||
测试环境当前可用：
|
||||
|
||||
| 字段 | 值 |
|
||||
| --- | --- |
|
||||
| customerCode | `DW0100046361` |
|
||||
| customerName | `重庆亿昊药业有限公司` |
|
||||
|
||||
联调入参用：
|
||||
|
||||
- `examples/contract_create_input_dry_run.json`：真客商 + `dry_run`，先校准页面
|
||||
- `examples/contract_draft_yyc.json`：医药城，同上客商
|
||||
- `examples/contract_draft_xyy.json`：小药药分支（不上传 PDF），客商仍用合同系统里能搜到的同一条
|
||||
|
||||
YYC 有多份平台 PDF 时需要合并，先安装：
|
||||
|
||||
|
|
@ -75,7 +85,7 @@ python -m rpa_yisaiqi.contract_create_node `
|
|||
--state-center-url http://192.168.x.x:8000 `
|
||||
--token dev-agent-token `
|
||||
--job-id <状态中心job_id> `
|
||||
--supplier-name 重庆亿昊药业 `
|
||||
--supplier-name 重庆亿昊药业有限公司 `
|
||||
--order-date 2026-08-11 `
|
||||
--source-type YYC `
|
||||
--draft-path rpa_yisaiqi/examples/contract_draft_yyc.json `
|
||||
|
|
@ -111,6 +121,27 @@ python -m rpa_yisaiqi.contract_create_node `
|
|||
--state-center-url http://192.168.x.x:8000
|
||||
```
|
||||
|
||||
## 艺赛旗画布怎么看
|
||||
|
||||
Studio 里「插入代码块」标题改不了，会一直显示「插入代码块#1」。步骤名写在节点**备注**里，画布上出现在代码块下方，例如 `S06 填写签约方`。
|
||||
|
||||
流程按四段换行，不要再排成一条长龙：
|
||||
|
||||
```
|
||||
准备 开始 → S01 初始化 → S02 打开系统 → S03 登录
|
||||
进入起草 S04 进入起草 → S05 选择模板
|
||||
填写表单 S06 签约方 → S07 基础信息 → S08 金额日期 → S09 货品 → S10 付款
|
||||
提交回写 S11 上传PDF → S12 提交 → S13 读合同号 → S14 回调
|
||||
```
|
||||
|
||||
改步骤后先关掉 Studio，再生成画布：
|
||||
|
||||
```powershell
|
||||
python rpa_yisaiqi/build_multinode_seq.py
|
||||
```
|
||||
|
||||
然后重新打开 `rpa_yisaiqi/studio` 项目并编译。
|
||||
|
||||
## 接入真实页面自动化
|
||||
|
||||
真实艺赛旗页面操作按流程图拆在：
|
||||
|
|
|
|||
|
|
@ -0,0 +1,17 @@
|
|||
# coding=utf-8
|
||||
"""多节点流程图（Main.seq）的跨节点共享状态容器。
|
||||
|
||||
艺赛旗编译 Main.seq 后，各代码节点的 PBCode 会在同一 Python 进程中执行。
|
||||
本模块以「模块级全局变量」承载 browser / ctx / client / spec 等对象，
|
||||
避免依赖艺赛旗对局部变量的拼接方式，让每个节点都能可靠地取到上一步的产物。
|
||||
"""
|
||||
|
||||
spec = None
|
||||
client = None
|
||||
ctx = None
|
||||
browser = None
|
||||
submitted = False
|
||||
contract_no = ""
|
||||
fd_id = ""
|
||||
is_dry_run = False
|
||||
completed = False
|
||||
|
|
@ -3,15 +3,43 @@
|
|||
APA (艺赛旗) Selenium Chrome BrowserAdapter 实现
|
||||
通过 Selenium WebDriver 直接操控 Chrome，不依赖 APA 浏览器模块
|
||||
"""
|
||||
from __future__ import annotations
|
||||
|
||||
import os
|
||||
import time
|
||||
from pathlib import Path
|
||||
|
||||
REPO_ROOT = Path(__file__).resolve().parents[1]
|
||||
|
||||
|
||||
def _is_inside_repo(path: Path, repo_root: Path) -> bool:
|
||||
try:
|
||||
path.resolve().relative_to(repo_root.resolve())
|
||||
except ValueError:
|
||||
return False
|
||||
return True
|
||||
|
||||
|
||||
def resolve_chrome_profile_base(requested: str | None = None, *, repo_root: Path | None = None) -> Path:
|
||||
"""Chrome user-data dir must stay outside the repo so list_dir never walks 3GB."""
|
||||
root = (repo_root or REPO_ROOT).resolve()
|
||||
outside = (root.parent / "rpa-chrome-profile").resolve()
|
||||
raw = os.getenv("RPA_CHROME_PROFILE") if requested is None else requested
|
||||
if raw:
|
||||
candidate = Path(raw).expanduser().resolve()
|
||||
if _is_inside_repo(candidate, root):
|
||||
print(f"RPA_CHROME_PROFILE 指向仓库内 {candidate}，已改用 {outside}", flush=True)
|
||||
return outside
|
||||
return candidate
|
||||
return outside
|
||||
|
||||
|
||||
class ApaBrowserAdapter:
|
||||
"""Selenium Chrome 浏览器适配器，实现 BrowserAdapter 协议"""
|
||||
|
||||
def __init__(self, url: str = ""):
|
||||
def __init__(self, url: str = "", *, flow: str = "default"):
|
||||
self._url = url
|
||||
self._flow = flow
|
||||
self._driver = None
|
||||
self._is_open = False
|
||||
|
||||
|
|
@ -23,21 +51,41 @@ class ApaBrowserAdapter:
|
|||
from selenium import webdriver
|
||||
from selenium.webdriver.chrome.service import Service
|
||||
from selenium.webdriver.chrome.options import Options
|
||||
from webdriver_manager.chrome import ChromeDriverManager
|
||||
|
||||
profile_base = resolve_chrome_profile_base()
|
||||
if _is_inside_repo(profile_base, REPO_ROOT):
|
||||
raise RuntimeError(f"拒绝在仓库内创建 Chrome 用户目录: {profile_base}")
|
||||
# 每次进程单独目录：上次 detach 留下的 Chrome 会锁住同一 profile
|
||||
profile_dir = profile_base / str(os.getpid())
|
||||
os.makedirs(profile_dir, exist_ok=True)
|
||||
profile_dir = str(profile_dir)
|
||||
|
||||
options = Options()
|
||||
options.page_load_strategy = "eager"
|
||||
# 不自动关闭浏览器，方便调试
|
||||
options.add_experimental_option("detach", True)
|
||||
# 避免被检测为自动化
|
||||
options.add_experimental_option("excludeSwitches", ["enable-automation"])
|
||||
options.add_experimental_option("excludeSwitches", ["enable-automation", "enable-logging"])
|
||||
options.add_experimental_option("useAutomationExtension", False)
|
||||
# 基本参数
|
||||
options.add_argument("--disable-blink-features=AutomationControlled")
|
||||
options.add_argument("--no-first-run")
|
||||
options.add_argument("--no-default-browser-check")
|
||||
options.add_argument("--disable-gpu")
|
||||
options.add_argument("--disable-extensions")
|
||||
options.add_argument("--disable-popup-blocking")
|
||||
options.add_argument("--start-maximized")
|
||||
# 独立目录，避免和日常 Chrome 抢用户数据目录导致卡住
|
||||
options.add_argument("--user-data-dir=" + profile_dir)
|
||||
|
||||
chromedriver = os.getenv("CHROMEDRIVER_PATH")
|
||||
if chromedriver and os.path.exists(chromedriver):
|
||||
service = Service(chromedriver)
|
||||
else:
|
||||
from webdriver_manager.chrome import ChromeDriverManager
|
||||
service = Service(ChromeDriverManager().install())
|
||||
self._driver = webdriver.Chrome(service=service, options=options)
|
||||
self._driver.set_page_load_timeout(45)
|
||||
self._driver.set_script_timeout(30)
|
||||
return self._driver
|
||||
|
||||
def _ensure_open(self):
|
||||
|
|
@ -110,7 +158,18 @@ class ApaBrowserAdapter:
|
|||
|
||||
def open_url(self, url: str) -> None:
|
||||
self._url = url
|
||||
self._get_driver().get(url)
|
||||
driver = self._get_driver()
|
||||
try:
|
||||
driver.get(url)
|
||||
except Exception as exc:
|
||||
# eager 仍可能因长轮询资源超时；页面文档已出来就继续
|
||||
current = ""
|
||||
try:
|
||||
current = driver.current_url or ""
|
||||
except Exception:
|
||||
current = ""
|
||||
if not current or current in ("data:,", "about:blank"):
|
||||
raise RuntimeError(f"打开页面失败：{url} ({exc})") from exc
|
||||
self._is_open = True
|
||||
|
||||
def _set_value_via_js(self, el, value: str) -> None:
|
||||
|
|
@ -153,6 +212,27 @@ class ApaBrowserAdapter:
|
|||
continue
|
||||
raise RuntimeError(f"fill_first failed: {selectors}, last: {last_error}")
|
||||
|
||||
def click_native(self, selectors: list) -> None:
|
||||
"""真实鼠标点击，给只监听 mousedown 的 LUI 页签/工具栏用。"""
|
||||
self._ensure_open()
|
||||
from selenium.webdriver.common.action_chains import ActionChains
|
||||
last_error = None
|
||||
for sel in selectors:
|
||||
try:
|
||||
el = self._find(sel)
|
||||
self._driver.execute_script(
|
||||
"arguments[0].scrollIntoView({block:'center', inline:'center'});",
|
||||
el,
|
||||
)
|
||||
time.sleep(0.2)
|
||||
ActionChains(self._driver).move_to_element(el).pause(0.15).click().perform()
|
||||
time.sleep(0.4)
|
||||
return
|
||||
except Exception as e:
|
||||
last_error = e
|
||||
continue
|
||||
raise RuntimeError(f"click_native failed: {selectors}, last: {last_error}")
|
||||
|
||||
def click_first(self, selectors: list) -> None:
|
||||
self._ensure_open()
|
||||
last_error = None
|
||||
|
|
@ -220,17 +300,34 @@ class ApaBrowserAdapter:
|
|||
return False
|
||||
|
||||
def switch_to_frame(self, selectors: list) -> None:
|
||||
"""切换到匹配到的 iframe（客商选择弹窗等）。"""
|
||||
"""切换到匹配到的 iframe。优先可见弹窗，避免切到上次留下的隐藏 iframe。"""
|
||||
self._ensure_open()
|
||||
from selenium.webdriver.common.by import By
|
||||
|
||||
last_error = None
|
||||
found = []
|
||||
for sel in selectors:
|
||||
try:
|
||||
el = self._find(sel, timeout=1)
|
||||
if sel.startswith("//") or sel.startswith("/"):
|
||||
found.extend(self._driver.find_elements(By.XPATH, sel))
|
||||
else:
|
||||
found.extend(self._driver.find_elements(By.CSS_SELECTOR, sel))
|
||||
except Exception as e:
|
||||
last_error = e
|
||||
visible = []
|
||||
hidden = []
|
||||
for el in found:
|
||||
try:
|
||||
(visible if el.is_displayed() else hidden).append(el)
|
||||
except Exception:
|
||||
hidden.append(el)
|
||||
for el in visible + hidden:
|
||||
try:
|
||||
self._driver.switch_to.default_content()
|
||||
self._driver.switch_to.frame(el)
|
||||
return
|
||||
except Exception as e:
|
||||
last_error = e
|
||||
continue
|
||||
raise RuntimeError(f"switch_to_frame failed: {selectors}, last: {last_error}")
|
||||
|
||||
def switch_to_default_content(self) -> None:
|
||||
|
|
@ -238,6 +335,70 @@ class ApaBrowserAdapter:
|
|||
self._ensure_open()
|
||||
self._driver.switch_to.default_content()
|
||||
|
||||
def execute_script(self, script: str, *args):
|
||||
self._ensure_open()
|
||||
return self._driver.execute_script(script, *args)
|
||||
|
||||
def select_xform(self, name: str, value: str) -> None:
|
||||
"""选中 Landray xform fancy-select（.fs-wrap / .fs-option）。"""
|
||||
self._ensure_open()
|
||||
result = self._driver.execute_script(
|
||||
"""
|
||||
const name = arguments[0];
|
||||
const wanted = String(arguments[1] || '').trim();
|
||||
const flag = document.querySelector('xformflag[property="' + name + '"]');
|
||||
if (!flag) {
|
||||
return {ok: false, error: 'flag not found: ' + name};
|
||||
}
|
||||
const norm = (s) => String(s || '').replace(/\\s+/g, '').trim();
|
||||
const wantedN = norm(wanted);
|
||||
const opts = Array.from(flag.querySelectorAll('.fs-option'));
|
||||
let target = opts.find((o) => {
|
||||
const text = norm((o.querySelector('.fs-option-label') || o).textContent);
|
||||
const val = norm(o.getAttribute('data-value'));
|
||||
return text === wantedN || val === wantedN;
|
||||
});
|
||||
if (!target && wantedN.length >= 2) {
|
||||
target = opts.find((o) => {
|
||||
const text = norm((o.querySelector('.fs-option-label') || o).textContent);
|
||||
return text.indexOf(wantedN) >= 0;
|
||||
});
|
||||
}
|
||||
if (!target) {
|
||||
return {
|
||||
ok: false,
|
||||
error: 'option not found: ' + wanted,
|
||||
available: opts.map((o) => ((o.querySelector('.fs-option-label') || o).textContent || '').trim()),
|
||||
};
|
||||
}
|
||||
const labelWrap = flag.querySelector('.fs-label-wrap');
|
||||
if (labelWrap) { labelWrap.click(); }
|
||||
target.click();
|
||||
const input = flag.querySelector('input[name="' + name + '"]');
|
||||
const selected = target.getAttribute('data-value') || '';
|
||||
if (input) {
|
||||
const proto = window.HTMLInputElement.prototype;
|
||||
const setter = Object.getOwnPropertyDescriptor(proto, 'value').set;
|
||||
setter.call(input, selected);
|
||||
input.dispatchEvent(new Event('input', { bubbles: true }));
|
||||
input.dispatchEvent(new Event('change', { bubbles: true }));
|
||||
if (typeof __xformDispatch === 'function') {
|
||||
try { __xformDispatch(selected, input); } catch (e) {}
|
||||
}
|
||||
}
|
||||
return {
|
||||
ok: true,
|
||||
value: selected,
|
||||
label: ((flag.querySelector('.fs-label') || {}).textContent || '').trim(),
|
||||
};
|
||||
""",
|
||||
name,
|
||||
str(value),
|
||||
)
|
||||
if not result or not result.get("ok"):
|
||||
raise RuntimeError(f"select_xform failed: {name}={value} {result}")
|
||||
time.sleep(0.3)
|
||||
|
||||
def select_first(self, selectors: list, value: str) -> None:
|
||||
self._ensure_open()
|
||||
from selenium.webdriver.support.ui import Select
|
||||
|
|
@ -334,10 +495,11 @@ class ApaBrowserAdapter:
|
|||
return ""
|
||||
|
||||
def screenshot(self, name: str) -> str:
|
||||
import os
|
||||
save_dir = os.path.join(os.path.dirname(__file__), "screenshots")
|
||||
os.makedirs(save_dir, exist_ok=True)
|
||||
path = os.path.join(save_dir, f"{name}.png")
|
||||
from .runtime.accept import artifact_dir
|
||||
|
||||
save_dir = artifact_dir(self._flow)
|
||||
save_dir.mkdir(parents=True, exist_ok=True)
|
||||
path = save_dir / f"{name}.png"
|
||||
if self._driver:
|
||||
self._driver.save_screenshot(path)
|
||||
return path
|
||||
self._driver.save_screenshot(str(path))
|
||||
return str(path)
|
||||
|
|
|
|||
|
|
@ -2,12 +2,13 @@
|
|||
from __future__ import annotations
|
||||
|
||||
import json
|
||||
from typing import Any, Callable
|
||||
from typing import Any, Callable, Dict
|
||||
|
||||
from urllib.error import HTTPError
|
||||
from urllib.parse import quote
|
||||
from urllib.request import Request, urlopen
|
||||
|
||||
JsonDict = dict[str, Any]
|
||||
JsonDict = Dict[str, Any]
|
||||
Opener = Callable[[Request, float], Any]
|
||||
|
||||
|
||||
|
|
|
|||
|
|
@ -4,13 +4,19 @@ from __future__ import annotations
|
|||
import argparse
|
||||
import json
|
||||
import os
|
||||
import sys
|
||||
from dataclasses import dataclass, replace
|
||||
from pathlib import Path
|
||||
from typing import Any, Callable
|
||||
|
||||
from .apa_browser_adapter import ApaBrowserAdapter
|
||||
from .callback_client import StateCenterClient
|
||||
from .contract_draft import ContractDraft, contract_draft_from_dict, load_contract_draft
|
||||
from .contract_draft import (
|
||||
ContractDraft,
|
||||
contract_draft_from_dict,
|
||||
is_placeholder_customer_code,
|
||||
load_contract_draft,
|
||||
)
|
||||
from .contract_flow_steps import ContractFlowContext, run_contract_create_flow
|
||||
from .pdf_utils import ensure_merged_pdf
|
||||
from .runtime.errors import NodeError, PdfMergeUnsupported, UnknownRpaError
|
||||
|
|
@ -23,6 +29,39 @@ CONTRACT_URL = "http://contract-test.gzmpc.com:8080/"
|
|||
class ContractCreateError(NodeError):
|
||||
"""入参缺失 / 页面元素找不到 / 页面改版，均映射 UI_ELEMENT_NOT_FOUND。"""
|
||||
|
||||
def __init__(self, message: str = "") -> None:
|
||||
# Studio 编译出来的旧 S01 在 input_arg 为空时会直接 raise。
|
||||
# 这里拦下来改走默认联调 JSON，避免每次改 Main.py 都被覆盖。
|
||||
if _should_bootstrap_default_input(message):
|
||||
_bootstrap_default_input_and_exit()
|
||||
super().__init__(message)
|
||||
|
||||
|
||||
def default_contract_input_path() -> str:
|
||||
return str(
|
||||
Path(__file__).resolve().parent / "examples" / "contract_create_input_dry_run.json"
|
||||
)
|
||||
|
||||
|
||||
def _should_bootstrap_default_input(message: str) -> bool:
|
||||
if "缺少业务入参" not in str(message):
|
||||
return False
|
||||
script = os.path.basename(str(sys.argv[0] if sys.argv else "")).lower()
|
||||
return script == "main.py"
|
||||
|
||||
|
||||
def _bootstrap_default_input_and_exit() -> None:
|
||||
path = default_contract_input_path()
|
||||
if not Path(path).is_file():
|
||||
return
|
||||
print("未传入 -i/--input，使用默认联调入参：", path, flush=True)
|
||||
codes_dir = Path(__file__).resolve().parent / "studio" / "codes"
|
||||
if str(codes_dir) not in sys.path:
|
||||
sys.path.insert(0, str(codes_dir))
|
||||
from run_contract import main
|
||||
|
||||
raise SystemExit(main())
|
||||
|
||||
|
||||
@dataclass(frozen=True)
|
||||
class ContractCreateInput:
|
||||
|
|
@ -95,7 +134,7 @@ def create_contract_in_contract_system(spec: ContractCreateInput) -> str:
|
|||
raise ContractCreateError("缺少 flow_context 或 draft_path，无法起草合同")
|
||||
ctx = build_flow_context(spec, load_contract_draft(spec.draft_path))
|
||||
|
||||
browser = ApaBrowserAdapter(url=ctx.contract_url)
|
||||
browser = ApaBrowserAdapter(url=ctx.contract_url, flow="CONTRACT_CREATE")
|
||||
return run_contract_create_flow(browser, ctx)
|
||||
|
||||
|
||||
|
|
@ -141,7 +180,9 @@ def build_flow_context(spec: ContractCreateInput, draft: ContractDraft) -> Contr
|
|||
contract_background=draft.contract_background,
|
||||
contract_nature=draft.contract_nature,
|
||||
amount_currency=draft.amount_currency,
|
||||
amount_type=draft.amount_type,
|
||||
amount=draft.amount,
|
||||
contract_title=draft.contract_title,
|
||||
contract_end_date=draft.contract_end_date,
|
||||
is_major_contract=draft.is_major_contract,
|
||||
is_prepayment=draft.is_prepayment,
|
||||
|
|
@ -153,6 +194,7 @@ def build_flow_context(spec: ContractCreateInput, draft: ContractDraft) -> Contr
|
|||
submit_mode=draft.submit_mode,
|
||||
goods_lines=draft.goods_lines,
|
||||
merged_pdf_path=merged_pdf_path,
|
||||
contract_document_path=draft.contract_document_path,
|
||||
order_no_list=draft.order_no_list,
|
||||
)
|
||||
|
||||
|
|
@ -166,6 +208,8 @@ def _resolve_merged_pdf(draft: ContractDraft, spec: ContractCreateInput) -> str
|
|||
|
||||
paths = [str(p).strip() for p in draft.platform_pdf_paths if str(p).strip()]
|
||||
if not paths:
|
||||
if draft.contract_document_path:
|
||||
return None
|
||||
raise PdfMergeUnsupported("YYC 缺少平台合同 PDF（platformPdfPaths 为空）")
|
||||
for path in paths:
|
||||
if not Path(path).exists():
|
||||
|
|
@ -200,6 +244,13 @@ def _validate_data(data: dict[str, Any], draft: ContractDraft) -> None:
|
|||
if not _str(data.get(key)):
|
||||
missing.append(key)
|
||||
|
||||
customer_code = draft.customer_code or _str(data.get("customerCode"))
|
||||
if customer_code and is_placeholder_customer_code(customer_code):
|
||||
raise ContractCreateError(
|
||||
f"customerCode「{customer_code}」是占位符，合同系统客商选择窗搜不到。"
|
||||
"请换成测试环境真实存在的客商操作码，例如 DW0100046361（重庆亿昊药业有限公司）"
|
||||
)
|
||||
|
||||
if draft.submit_mode not in ("dry_run", "submit"):
|
||||
raise ContractCreateError(f"submitMode 非法：{draft.submit_mode}（仅支持 dry_run / submit）")
|
||||
|
||||
|
|
@ -252,6 +303,11 @@ def load_input_data(raw: str) -> dict[str, Any]:
|
|||
"""解析入参：优先按 JSON 字符串，其次按 JSON 文件路径。"""
|
||||
raw = (raw or "").strip()
|
||||
if not raw:
|
||||
fallback = default_contract_input_path()
|
||||
if Path(fallback).is_file():
|
||||
print("未传入 -i/--input，使用默认联调入参：", fallback, flush=True)
|
||||
raw = fallback
|
||||
else:
|
||||
return {}
|
||||
try:
|
||||
obj = json.loads(raw)
|
||||
|
|
|
|||
|
|
@ -8,6 +8,14 @@ from typing import Any
|
|||
|
||||
from .contract_flow_steps import ContractGoodsLine
|
||||
|
||||
_PLACEHOLDER_CUSTOMER_PREFIXES = ("CUST-", "TEST-", "MOCK-")
|
||||
|
||||
|
||||
def is_placeholder_customer_code(code: str) -> bool:
|
||||
"""CUST-001 / TEST-xxx 这类占位符在合同系统里搜不到，不能拿去点真页面。"""
|
||||
value = (code or "").strip().upper()
|
||||
return value.startswith(_PLACEHOLDER_CUSTOMER_PREFIXES)
|
||||
|
||||
|
||||
@dataclass(frozen=True)
|
||||
class ContractDraft:
|
||||
|
|
@ -23,7 +31,9 @@ class ContractDraft:
|
|||
contract_background: str = ""
|
||||
contract_nature: str = "主合同"
|
||||
amount_currency: str = "人民币"
|
||||
amount_type: str = "合同标的金额"
|
||||
amount: str = ""
|
||||
contract_title: str = ""
|
||||
contract_end_date: str = ""
|
||||
is_major_contract: str = "否"
|
||||
is_prepayment: str = "否"
|
||||
|
|
@ -36,6 +46,7 @@ class ContractDraft:
|
|||
goods_lines: list[ContractGoodsLine] = field(default_factory=list)
|
||||
platform_pdf_paths: list[str] = field(default_factory=list)
|
||||
merged_pdf_path: str | None = None
|
||||
contract_document_path: str = ""
|
||||
order_no_list: list[str] = field(default_factory=list)
|
||||
|
||||
|
||||
|
|
@ -72,7 +83,9 @@ def contract_draft_from_dict(data: dict[str, Any]) -> ContractDraft:
|
|||
contract_background=str(data.get("contractBackground", "")),
|
||||
contract_nature=str(data.get("contractNature", "主合同")),
|
||||
amount_currency=str(data.get("amountCurrency", "人民币")),
|
||||
amount_type=str(data.get("amountType", "合同标的金额")),
|
||||
amount=str(data.get("amount", "")),
|
||||
contract_title=str(data.get("contractTitle") or data.get("docSubject") or ""),
|
||||
contract_end_date=str(data.get("contractEndDate", "")),
|
||||
is_major_contract=str(data.get("isMajorContract", "否")),
|
||||
is_prepayment=str(data.get("isPrepayment", "否")),
|
||||
|
|
@ -85,5 +98,8 @@ def contract_draft_from_dict(data: dict[str, Any]) -> ContractDraft:
|
|||
goods_lines=goods_lines,
|
||||
platform_pdf_paths=[str(path) for path in data.get("platformPdfPaths", [])],
|
||||
merged_pdf_path=data.get("mergedPdfPath"),
|
||||
contract_document_path=str(
|
||||
data.get("contractDocumentPath") or data.get("contractBodyPath") or ""
|
||||
),
|
||||
order_no_list=[str(x) for x in data.get("orderNoList", [])],
|
||||
)
|
||||
|
|
|
|||
File diff suppressed because it is too large
Load Diff
|
|
@ -1,5 +1,5 @@
|
|||
{
|
||||
"jobId": "TEST-CONTRACT-20260813-001",
|
||||
"jobId": "YYC-重庆亿昊药业-2fa0-20260812-16abff",
|
||||
"supplierName": "重庆亿昊药业有限公司",
|
||||
"orderDate": "2026-08-12",
|
||||
"sourceType": "YYC",
|
||||
|
|
@ -7,17 +7,19 @@
|
|||
"templateName": "非格式--采购协议",
|
||||
"customerCode": "DW0100046361",
|
||||
"customerName": "重庆亿昊药业有限公司",
|
||||
"contractCategory": "采购协议",
|
||||
"contractCategory": "其他协议（采购）",
|
||||
"settlementType": "付款",
|
||||
"contractBackground": "自动化测试合同起草",
|
||||
"contractNature": "主合同",
|
||||
"contractTitle": "test",
|
||||
"amountCurrency": "人民币",
|
||||
"amountType": "合同标的金额",
|
||||
"amount": "50000.00",
|
||||
"contractEndDate": "2026-12-31",
|
||||
"isMajorContract": "否",
|
||||
"isPrepayment": "否",
|
||||
"belongYear": "2026",
|
||||
"paymentMethod": "账期支付",
|
||||
"paymentMethod": "电汇",
|
||||
"paymentDays": "30",
|
||||
"contractStatus": "正式合同",
|
||||
"containsBloodProducts": "否",
|
||||
|
|
@ -27,14 +29,14 @@
|
|||
"goodsLines": [
|
||||
{
|
||||
"operationCode": "ADD",
|
||||
"goodsCode": "YP-001",
|
||||
"goodsName": "测试药品A",
|
||||
"approvalNo": "国药准字H00000001",
|
||||
"spec": "50mg*30片",
|
||||
"goodsCode": "FQX0100118047",
|
||||
"goodsName": "盐酸托烷司琼注射液",
|
||||
"approvalNo": "苏20160372",
|
||||
"spec": "5ml:5mg*5支",
|
||||
"unit": "盒",
|
||||
"quantity": 10,
|
||||
"mdmCode": "",
|
||||
"unitPrice": ""
|
||||
"unitPrice": "50.00"
|
||||
}
|
||||
],
|
||||
"stateCenterUrl": "http://127.0.0.1:8000",
|
||||
|
|
|
|||
|
|
@ -1,11 +1,11 @@
|
|||
{
|
||||
"supplierName": "小药药供应商",
|
||||
"supplierName": "重庆亿昊药业有限公司",
|
||||
"orderDate": "2026-08-11",
|
||||
"sourceType": "XYY",
|
||||
"templateType": "FORMAT",
|
||||
"templateName": "购销协议书（2025）",
|
||||
"customerCode": "CUST-XYY-001",
|
||||
"customerName": "小药药供应商",
|
||||
"customerCode": "DW0100046361",
|
||||
"customerName": "重庆亿昊药业有限公司",
|
||||
"contractCategory": "采购协议",
|
||||
"settlementType": "付款",
|
||||
"contractBackground": "按小药药订单自动起草采购协议",
|
||||
|
|
|
|||
|
|
@ -1,11 +1,11 @@
|
|||
{
|
||||
"supplierName": "重庆亿昊药业",
|
||||
"supplierName": "重庆亿昊药业有限公司",
|
||||
"orderDate": "2026-08-11",
|
||||
"sourceType": "YYC",
|
||||
"templateType": "NON_FORMAT",
|
||||
"templateName": "非格式--采购协议",
|
||||
"customerCode": "CUST-YYC-001",
|
||||
"customerName": "重庆亿昊药业",
|
||||
"customerCode": "DW0100046361",
|
||||
"customerName": "重庆亿昊药业有限公司",
|
||||
"contractCategory": "采购协议",
|
||||
"settlementType": "付款",
|
||||
"contractBackground": "按医药城订单自动起草采购协议",
|
||||
|
|
|
|||
|
|
@ -0,0 +1,153 @@
|
|||
"""Shared ACCEPT report for every RPA flow (any platform / page).
|
||||
|
||||
New flows (ORDER_FETCH, REBATE_*, M5, ...) must use this instead of
|
||||
reading screenshots. Screenshots stay on disk for humans; this text is
|
||||
the agent oracle.
|
||||
"""
|
||||
from __future__ import annotations
|
||||
|
||||
import os
|
||||
from dataclasses import dataclass, field
|
||||
from pathlib import Path
|
||||
from typing import Protocol
|
||||
|
||||
from .errors import NodeError
|
||||
|
||||
PACKAGE_ROOT = Path(__file__).resolve().parents[1]
|
||||
SCREENSHOT_ROOT = PACKAGE_ROOT / "screenshots"
|
||||
|
||||
|
||||
class HasScreenshot(Protocol):
|
||||
def screenshot(self, name: str) -> str: ...
|
||||
|
||||
|
||||
@dataclass
|
||||
class AcceptLine:
|
||||
step: str
|
||||
title: str
|
||||
ok: bool
|
||||
facts: dict[str, str]
|
||||
mismatches: list[str]
|
||||
|
||||
|
||||
def safe_flow_name(flow: str) -> str:
|
||||
text = (flow or "default").strip() or "default"
|
||||
return "".join(ch if ch.isalnum() or ch in "-_" else "_" for ch in text)
|
||||
|
||||
|
||||
def artifact_dir(flow: str) -> Path:
|
||||
return SCREENSHOT_ROOT / safe_flow_name(flow)
|
||||
|
||||
|
||||
def evidence_shot(browser: HasScreenshot, name: str, *, failed: bool = False) -> str:
|
||||
"""Save a screenshot only on failure. Set RPA_SCREENSHOT=always to keep every frame."""
|
||||
mode = (os.environ.get("RPA_SCREENSHOT") or "fail").strip().lower()
|
||||
if not failed and mode != "always":
|
||||
return ""
|
||||
return browser.screenshot(name)
|
||||
|
||||
|
||||
def accept_log_path(flow: str = "default") -> Path:
|
||||
override = os.environ.get("RPA_ACCEPT_LOG")
|
||||
if override:
|
||||
return Path(override)
|
||||
return artifact_dir(flow) / "ACCEPT.txt"
|
||||
|
||||
|
||||
def field_match(actual: str, expected: str) -> bool:
|
||||
actual_text = (actual or "").strip()
|
||||
expected_text = (expected or "").strip()
|
||||
if not expected_text:
|
||||
return True
|
||||
if actual_text == expected_text:
|
||||
return True
|
||||
compact_a = actual_text.replace(" ", "").replace(",", "")
|
||||
compact_e = expected_text.replace(" ", "").replace(",", "")
|
||||
if compact_a == compact_e or compact_a.replace("-", "") == compact_e.replace("-", ""):
|
||||
return True
|
||||
try:
|
||||
return float(compact_a) == float(compact_e)
|
||||
except ValueError:
|
||||
pass
|
||||
yes = {"是", "1", "Y", "YES", "TRUE"}
|
||||
no = {"否", "2", "0", "N", "NO", "FALSE"}
|
||||
if compact_a.upper() in yes and compact_e.upper() in yes:
|
||||
return True
|
||||
if compact_a.upper() in no and compact_e.upper() in no:
|
||||
return True
|
||||
return compact_e in compact_a or compact_a in compact_e
|
||||
|
||||
|
||||
def format_accept(line: AcceptLine) -> str:
|
||||
status = "ok" if line.ok else "FAIL"
|
||||
facts = " ".join(f"{key}={value}" for key, value in line.facts.items())
|
||||
extra = f" {' | '.join(line.mismatches)}" if line.mismatches else ""
|
||||
return f"ACCEPT {line.step} {status} {line.title} {facts}{extra}".rstrip()
|
||||
|
||||
|
||||
@dataclass
|
||||
class AcceptanceReport:
|
||||
"""DOM / page-state checklist. One instance per flow run."""
|
||||
|
||||
flow: str = "default"
|
||||
lines: list[AcceptLine] = field(default_factory=list)
|
||||
|
||||
def reset(self) -> None:
|
||||
self.lines = []
|
||||
|
||||
def check(
|
||||
self,
|
||||
step: str,
|
||||
title: str,
|
||||
pairs: list[tuple[str, object, object]],
|
||||
*,
|
||||
browser: HasScreenshot | None = None,
|
||||
screenshot_on_fail: str = "",
|
||||
) -> None:
|
||||
facts: dict[str, str] = {}
|
||||
mismatches: list[str] = []
|
||||
for name, actual, expected in pairs:
|
||||
actual_text = "" if actual is None else str(actual).strip()
|
||||
expected_text = "" if expected is None else str(expected).strip()
|
||||
facts[name] = actual_text or "(empty)"
|
||||
if expected_text and not field_match(actual_text, expected_text):
|
||||
mismatches.append(f"{name} expected={expected_text!r} actual={actual_text!r}")
|
||||
ok = not mismatches
|
||||
self.lines.append(AcceptLine(step, title, ok, facts, mismatches))
|
||||
print(format_accept(self.lines[-1]), flush=True)
|
||||
self.dump()
|
||||
if ok:
|
||||
return
|
||||
if browser is not None and screenshot_on_fail:
|
||||
evidence_shot(browser, screenshot_on_fail, failed=True)
|
||||
raise NodeError(f"ACCEPT {step} FAIL " + "; ".join(mismatches))
|
||||
|
||||
def note(
|
||||
self,
|
||||
step: str,
|
||||
title: str,
|
||||
facts: dict[str, object],
|
||||
*,
|
||||
ok: bool = True,
|
||||
raise_error: bool = True,
|
||||
) -> None:
|
||||
rendered = {key: ("" if value is None else str(value).strip()) or "(empty)" for key, value in facts.items()}
|
||||
self.lines.append(AcceptLine(step, title, ok, rendered, [] if ok else ["failed"]))
|
||||
print(format_accept(self.lines[-1]), flush=True)
|
||||
self.dump()
|
||||
if not ok and raise_error:
|
||||
raise NodeError(f"ACCEPT {step} FAIL {title} " + " ".join(f"{k}={v}" for k, v in rendered.items()))
|
||||
|
||||
def dump(self) -> None:
|
||||
path = accept_log_path(self.flow)
|
||||
try:
|
||||
path.parent.mkdir(parents=True, exist_ok=True)
|
||||
path.write_text(self.summary_text(), encoding="utf-8")
|
||||
except OSError as exc:
|
||||
print(f"ACCEPT log write skipped: {exc}", flush=True)
|
||||
|
||||
def summary_text(self) -> str:
|
||||
passed = sum(1 for line in self.lines if line.ok)
|
||||
rows = [f"ACCEPT SUMMARY {self.flow} {passed}/{len(self.lines)} ok"]
|
||||
rows.extend(format_accept(line) for line in self.lines)
|
||||
return "\n".join(rows) + "\n"
|
||||
|
|
@ -1,14 +1,13 @@
|
|||
"""Generic start → runner → complete/fail shell."""
|
||||
from __future__ import annotations
|
||||
|
||||
from collections.abc import Callable
|
||||
from typing import Any
|
||||
from typing import Any, Callable, Dict, Optional
|
||||
|
||||
from ..callback_client import StateCenterClient
|
||||
from .errors import NodeError
|
||||
from .payloads import build_complete_payload, build_fail_payload
|
||||
|
||||
Runner = Callable[[], dict[str, Any] | None]
|
||||
Runner = Callable[[], Optional[Dict[str, Any]]]
|
||||
|
||||
|
||||
def execute_node(
|
||||
|
|
|
|||
|
|
@ -12,11 +12,14 @@
|
|||
"form": {
|
||||
"customerCode": "input[name='extendDataFormInfo.value(fd_customeropcode)']",
|
||||
"customerName": "input[name='extendDataFormInfo.value(fd_customername)']",
|
||||
"contractCategory": "input[name='extendDataFormInfo.value(fd_contract_cate_type)']",
|
||||
"contractCategory": "xform:extendDataFormInfo.value(fd_contract_cate_type)",
|
||||
"coverTemplate": "xform:extendDataFormInfo.value(fd_contract_cover_type)",
|
||||
"settlementType": "select#fdSettype",
|
||||
"contractBackground": "textarea[name='extendDataFormInfo.value(fd_desc_info)']",
|
||||
"contractNatureMain": "input[name='fdType'][value='00']",
|
||||
"contractNatureSub": "input[name='fdType'][value='01']",
|
||||
"docSubject": "input[name='docSubject']",
|
||||
"amountType": "xform:extendDataFormInfo.value(fd_all_amount_type)",
|
||||
"amount": "input[name='fdAccount']",
|
||||
"signDate": "input[name='fdSigndate']",
|
||||
"beginDate": "input[name='fdEffedate']",
|
||||
|
|
@ -28,6 +31,8 @@
|
|||
"paymentDays": "input[name='extendDataFormInfo.value(fd_pay_time_limit)']",
|
||||
"contractStatus": "input[name='extendDataFormInfo.value(fd_suconstatus)']",
|
||||
"containsBloodProducts": "input[name='extendDataFormInfo.value(fd_3d1ce3953f1162)']",
|
||||
"selectGoodsButton": "div[title='选择商品']",
|
||||
"goodsDialogFrame": "iframe[src*='EopBasedataMaterial']",
|
||||
"goodsOperationCode": "input[name='extendDataFormInfo.value(fd_detail1.0.fd_opcode)']",
|
||||
"goodsCode": "input[name='extendDataFormInfo.value(fd_detail1.0.fd_opcode)']",
|
||||
"goodsName": "input[name='extendDataFormInfo.value(fd_detail1.0.fd_goodsname)']",
|
||||
|
|
|
|||
|
|
@ -0,0 +1,33 @@
|
|||
from pathlib import Path
|
||||
|
||||
from rpa_yisaiqi.apa_browser_adapter import REPO_ROOT, resolve_chrome_profile_base
|
||||
|
||||
|
||||
def test_default_chrome_profile_is_outside_repo():
|
||||
profile = resolve_chrome_profile_base(requested="")
|
||||
assert profile == (REPO_ROOT.parent / "rpa-chrome-profile").resolve()
|
||||
assert REPO_ROOT.resolve() not in profile.parents
|
||||
assert profile != REPO_ROOT.resolve()
|
||||
|
||||
|
||||
def test_env_path_inside_repo_is_rejected(tmp_path, monkeypatch):
|
||||
inside = REPO_ROOT / "rpa_yisaiqi" / "runtime" / "chrome-profile"
|
||||
monkeypatch.delenv("RPA_CHROME_PROFILE", raising=False)
|
||||
profile = resolve_chrome_profile_base(requested=str(inside))
|
||||
assert profile == (REPO_ROOT.parent / "rpa-chrome-profile").resolve()
|
||||
assert not _is_relative_to(profile, REPO_ROOT)
|
||||
|
||||
|
||||
def test_env_path_outside_repo_is_kept(tmp_path):
|
||||
outside = tmp_path / "custom-chrome"
|
||||
outside.mkdir()
|
||||
profile = resolve_chrome_profile_base(requested=str(outside))
|
||||
assert profile == outside.resolve()
|
||||
|
||||
|
||||
def _is_relative_to(path: Path, root: Path) -> bool:
|
||||
try:
|
||||
path.resolve().relative_to(root.resolve())
|
||||
except ValueError:
|
||||
return False
|
||||
return True
|
||||
|
|
@ -0,0 +1,59 @@
|
|||
from rpa_yisaiqi.runtime.accept import (
|
||||
AcceptanceReport,
|
||||
accept_log_path,
|
||||
artifact_dir,
|
||||
evidence_shot,
|
||||
field_match,
|
||||
)
|
||||
from rpa_yisaiqi.runtime.errors import NodeError
|
||||
import pytest
|
||||
|
||||
|
||||
def test_each_flow_writes_its_own_accept_file(tmp_path, monkeypatch):
|
||||
monkeypatch.delenv("RPA_ACCEPT_LOG", raising=False)
|
||||
monkeypatch.setattr(
|
||||
"rpa_yisaiqi.runtime.accept.SCREENSHOT_ROOT",
|
||||
tmp_path,
|
||||
)
|
||||
report = AcceptanceReport(flow="ORDER_FETCH")
|
||||
report.note("S01", "登录", {"user": "u1"}, ok=True)
|
||||
path = accept_log_path("ORDER_FETCH")
|
||||
assert path == tmp_path / "ORDER_FETCH" / "ACCEPT.txt"
|
||||
text = path.read_text(encoding="utf-8")
|
||||
assert "ACCEPT SUMMARY ORDER_FETCH 1/1 ok" in text
|
||||
assert "ACCEPT S01 ok 登录 user=u1" in text
|
||||
|
||||
|
||||
def test_artifact_dir_is_per_flow():
|
||||
assert artifact_dir("REBATE_HEADER").name == "REBATE_HEADER"
|
||||
assert artifact_dir("M5_CREATE_ORDER").name == "M5_CREATE_ORDER"
|
||||
assert artifact_dir("CONTRACT_CREATE") != artifact_dir("ORDER_FETCH")
|
||||
|
||||
|
||||
def test_check_fails_on_mismatch_without_reading_images():
|
||||
report = AcceptanceReport(flow="REBATE_HEADER")
|
||||
with pytest.raises(NodeError, match="ACCEPT S02 FAIL"):
|
||||
report.check("S02", "总单号", [("rebateOrderNo", "RB-1", "RB-2")])
|
||||
|
||||
|
||||
def test_evidence_shot_skips_success_and_keeps_failure():
|
||||
class Fake:
|
||||
def __init__(self):
|
||||
self.names = []
|
||||
|
||||
def screenshot(self, name):
|
||||
self.names.append(name)
|
||||
return name
|
||||
|
||||
browser = Fake()
|
||||
assert evidence_shot(browser, "ok") == ""
|
||||
assert browser.names == []
|
||||
assert evidence_shot(browser, "bad", failed=True) == "bad"
|
||||
assert browser.names == ["bad"]
|
||||
|
||||
|
||||
def test_field_match_accepts_numeric_and_yes_no():
|
||||
assert field_match("12.0", "12")
|
||||
assert field_match("是", "1")
|
||||
assert field_match("2026-08-11", "20260811")
|
||||
assert not field_match("重庆其他", "重庆亿昊药业有限公司")
|
||||
|
|
@ -37,6 +37,6 @@ class YiSaiQiBrowserAdapter:
|
|||
raise NotImplementedError(f"Map to YiSaiQi read-text activity: {selectors=} {value_name=}")
|
||||
|
||||
def screenshot(self, name: str) -> str:
|
||||
"""Flowchart node: 截图存证，返回文件路径."""
|
||||
"""Human evidence only. Acceptance is ACCEPT text from runtime.accept, not this image."""
|
||||
raise NotImplementedError(f"Map to YiSaiQi screenshot activity: {name=}")
|
||||
|
||||
|
|
|
|||
|
|
@ -3,13 +3,15 @@ from pathlib import Path
|
|||
|
||||
import pytest
|
||||
|
||||
from rpa_yisaiqi.apa_browser_adapter import ApaBrowserAdapter
|
||||
from rpa_yisaiqi.contract_create_node import (
|
||||
ContractCreateError,
|
||||
ContractCreateInput,
|
||||
PdfMergeUnsupported,
|
||||
build_spec_from_data,
|
||||
create_contract_in_contract_system,
|
||||
)
|
||||
from rpa_yisaiqi.contract_draft import load_contract_draft
|
||||
from rpa_yisaiqi.contract_draft import is_placeholder_customer_code, load_contract_draft
|
||||
from rpa_yisaiqi.pdf_utils import ensure_merged_pdf
|
||||
|
||||
|
||||
|
|
@ -77,6 +79,33 @@ def test_load_contract_draft_maps_full_contract_inputs(tmp_path):
|
|||
assert draft.goods_lines[0].goods_code == "YP-001"
|
||||
|
||||
|
||||
def test_placeholder_customer_code_is_rejected_before_opening_browser():
|
||||
assert is_placeholder_customer_code("CUST-YYC-001")
|
||||
assert is_placeholder_customer_code("test-001")
|
||||
assert not is_placeholder_customer_code("DW0100046361")
|
||||
|
||||
with pytest.raises(ContractCreateError, match="占位符"):
|
||||
build_spec_from_data(
|
||||
{
|
||||
"jobId": "JOB-1",
|
||||
"supplierName": "重庆亿昊药业有限公司",
|
||||
"orderDate": "2026-08-11",
|
||||
"sourceType": "XYY",
|
||||
"customerCode": "CUST-YYC-001",
|
||||
"customerName": "重庆亿昊药业有限公司",
|
||||
"amount": "100",
|
||||
"submitMode": "dry_run",
|
||||
"goodsLines": [
|
||||
{
|
||||
"goodsCode": "YP-001",
|
||||
"goodsName": "阿卡波糖片",
|
||||
"quantity": 1,
|
||||
}
|
||||
],
|
||||
}
|
||||
)
|
||||
|
||||
|
||||
def test_yyc_requires_pdf_inputs_before_contract_drafting(tmp_path):
|
||||
draft_path = tmp_path / "draft.json"
|
||||
draft_path.write_text(
|
||||
|
|
@ -111,14 +140,16 @@ def test_ensure_merged_pdf_uses_single_pdf_without_merge(tmp_path):
|
|||
assert ensure_merged_pdf([str(source)], str(tmp_path / "merged.pdf")) == str(source)
|
||||
|
||||
|
||||
def test_real_drafting_requires_yisaiqi_adapter_mapping(tmp_path):
|
||||
def test_create_contract_uses_apa_browser_adapter(tmp_path, monkeypatch):
|
||||
draft_path = tmp_path / "draft.json"
|
||||
draft_path.write_text(
|
||||
json.dumps(
|
||||
{
|
||||
"supplierName": "小药药供应商",
|
||||
"supplierName": "重庆亿昊药业有限公司",
|
||||
"orderDate": "2026-08-11",
|
||||
"sourceType": "XYY",
|
||||
"customerCode": "DW0100046361",
|
||||
"customerName": "重庆亿昊药业有限公司",
|
||||
"goodsLines": [],
|
||||
"platformPdfPaths": [],
|
||||
},
|
||||
|
|
@ -128,11 +159,22 @@ def test_real_drafting_requires_yisaiqi_adapter_mapping(tmp_path):
|
|||
)
|
||||
spec = ContractCreateInput(
|
||||
job_id="JOB-1",
|
||||
supplier_name="小药药供应商",
|
||||
supplier_name="重庆亿昊药业有限公司",
|
||||
order_date="2026-08-11",
|
||||
source_type="XYY",
|
||||
draft_path=str(draft_path),
|
||||
)
|
||||
captured = {}
|
||||
|
||||
with pytest.raises(ContractCreateError, match="YiSaiQiBrowserAdapter"):
|
||||
create_contract_in_contract_system(spec)
|
||||
def fake_run(browser, ctx):
|
||||
captured["browser"] = browser
|
||||
captured["ctx"] = ctx
|
||||
return "HT-1"
|
||||
|
||||
monkeypatch.setattr(
|
||||
"rpa_yisaiqi.contract_create_node.run_contract_create_flow",
|
||||
fake_run,
|
||||
)
|
||||
|
||||
assert create_contract_in_contract_system(spec) == "HT-1"
|
||||
assert isinstance(captured["browser"], ApaBrowserAdapter)
|
||||
|
|
|
|||
|
|
@ -1,110 +1,269 @@
|
|||
import pytest
|
||||
|
||||
from rpa_yisaiqi.contract_flow_steps import (
|
||||
ContractFlowContext,
|
||||
ContractGoodsLine,
|
||||
fill_contract_signatory_info,
|
||||
run_contract_create_flow,
|
||||
)
|
||||
from rpa_yisaiqi.runtime.errors import SupplierContractDisabled
|
||||
|
||||
|
||||
@pytest.fixture(autouse=True)
|
||||
def no_sleep(monkeypatch):
|
||||
monkeypatch.setattr("rpa_yisaiqi.contract_flow_steps.time.sleep", lambda _s: None)
|
||||
|
||||
|
||||
class FakeBrowser:
|
||||
def __init__(self):
|
||||
self.calls = []
|
||||
self.body_text = ""
|
||||
self.xform_labels = {}
|
||||
self.filled = {}
|
||||
self.read_values = {
|
||||
"contract_number": "HT-20260811-REAL-001",
|
||||
}
|
||||
|
||||
def open_url(self, url):
|
||||
self.calls.append(("open_url", url))
|
||||
if "kmAgreementApply" in url or "draft.jsp" in url:
|
||||
self.body_text = "客户编码 客户名称 合同类别"
|
||||
|
||||
def fill_first(self, selectors, value):
|
||||
self.calls.append(("fill_first", tuple(selectors), value))
|
||||
for selector in selectors:
|
||||
self.filled[selector] = value
|
||||
|
||||
def click_native(self, selectors):
|
||||
self.calls.append(("click_native", tuple(selectors)))
|
||||
|
||||
def click_first(self, selectors):
|
||||
self.calls.append(("click_first", tuple(selectors)))
|
||||
joined = " ".join(str(item) for item in selectors)
|
||||
if "login" in joined or "登录" in joined:
|
||||
self.body_text = "切换门户 待办 待阅 合同管理 合同起草"
|
||||
|
||||
def wait_for_text(self, text, timeout_seconds=15):
|
||||
self.calls.append(("wait_for_text", text, timeout_seconds))
|
||||
if text and text not in self.body_text:
|
||||
self.body_text = f"{self.body_text} {text}".strip()
|
||||
|
||||
def wait_for_element(self, selectors, timeout_seconds=15):
|
||||
self.calls.append(("wait_for_element", tuple(selectors), timeout_seconds))
|
||||
|
||||
def element_exists(self, selectors):
|
||||
self.calls.append(("element_exists", tuple(selectors)))
|
||||
return True
|
||||
|
||||
def switch_to_frame(self, selectors):
|
||||
self.calls.append(("switch_to_frame", tuple(selectors)))
|
||||
|
||||
def switch_to_default_content(self):
|
||||
self.calls.append(("switch_to_default_content",))
|
||||
|
||||
def select_first(self, selectors, value):
|
||||
self.calls.append(("select_first", tuple(selectors), value))
|
||||
for selector in selectors:
|
||||
self.filled[selector] = value
|
||||
|
||||
def select_xform(self, name, value):
|
||||
self.calls.append(("select_xform", name, value))
|
||||
self.xform_labels[name] = value
|
||||
|
||||
def execute_script(self, script, *args):
|
||||
self.calls.append(("execute_script", script[:48], args))
|
||||
if "hasTextInFrames" in script:
|
||||
return True
|
||||
if "latestByApplyTime" in script and "headerIdx" in script:
|
||||
return "001132026081734940"
|
||||
if "wantedTitle" in script and "headerIdx" in script:
|
||||
return "001132026081734940"
|
||||
if "fd_opcode" in script and "return i" in script:
|
||||
return 0
|
||||
if "List_Selected" in script and "innerText" in script:
|
||||
token = str(args[0]) if args else ""
|
||||
if token.startswith("DW"):
|
||||
self.filled['input[name="extendDataFormInfo.value(fd_customeropcode)"]'] = token
|
||||
elif token:
|
||||
self.filled['input[name="extendDataFormInfo.value(fd_detail1.0.fd_opcode)"]'] = token
|
||||
if "box.click" in script:
|
||||
return {"ok": True}
|
||||
return True
|
||||
if "marked" in script and "fd_detail1" in script:
|
||||
return []
|
||||
if "return location.href" in script:
|
||||
return "http://contract-test.gzmpc.com:8080/km/agreement/km_agreement_apply/kmAgreementApply.do?method=view"
|
||||
if "Com_Parameter" in script and "fdId" in script:
|
||||
return "FD-TEST-001"
|
||||
if "添加合同" in script and ("pdf" in script or "wantedStem" in script):
|
||||
if any(call[0] == "upload_file_first" for call in self.calls):
|
||||
return {"ok": True, "via": "name"}
|
||||
return {"ok": False, "via": "empty"}
|
||||
if "提交" in script or "合同正文" in script or "确定" in script:
|
||||
return {"ok": True}
|
||||
if "fs-label" in script and args:
|
||||
return self.xform_labels.get(args[0], "")
|
||||
if "data-value" in script:
|
||||
return "模板A"
|
||||
return []
|
||||
|
||||
def upload_file_first(self, selectors, file_path):
|
||||
self.calls.append(("upload_file_first", tuple(selectors), file_path))
|
||||
|
||||
def read_first(self, selectors, value_name):
|
||||
self.calls.append(("read_first", tuple(selectors), value_name))
|
||||
if value_name in self.read_values:
|
||||
return self.read_values[value_name]
|
||||
for selector in selectors:
|
||||
if selector in self.filled:
|
||||
return self.filled[selector]
|
||||
joined = " ".join(str(item) for item in selectors)
|
||||
if "fd_customername" in joined:
|
||||
return self.read_values.get("customer_name", "重庆亿昊药业有限公司")
|
||||
if "fd_customeropcode" in joined:
|
||||
return self.read_values.get("customer_code", self.filled.get(
|
||||
'input[name="extendDataFormInfo.value(fd_customeropcode)"]', ""
|
||||
))
|
||||
if "fdSettype" in joined:
|
||||
return self.read_values.get("settlement_type", "付款")
|
||||
return ""
|
||||
|
||||
def read_body_text(self):
|
||||
self.calls.append(("read_body_text",))
|
||||
return self.body_text
|
||||
|
||||
def screenshot(self, name):
|
||||
self.calls.append(("screenshot", name))
|
||||
return f"/tmp/{name}.png"
|
||||
|
||||
|
||||
def test_signatory_rejects_placeholder_customer_before_opening_picker():
|
||||
browser = FakeBrowser()
|
||||
ctx = ContractFlowContext(
|
||||
contract_url="http://contract-test.gzmpc.com:8080/",
|
||||
username="test002",
|
||||
password="1",
|
||||
supplier_name="重庆亿昊药业有限公司",
|
||||
order_date="2026-08-11",
|
||||
source_type="YYC",
|
||||
customer_code="CUST-YYC-001",
|
||||
customer_name="重庆亿昊药业有限公司",
|
||||
)
|
||||
|
||||
with pytest.raises(SupplierContractDisabled, match="占位符"):
|
||||
fill_contract_signatory_info(browser, ctx)
|
||||
|
||||
assert browser.calls == []
|
||||
|
||||
|
||||
def test_contract_flow_logs_in_with_real_contract_login_selectors():
|
||||
browser = FakeBrowser()
|
||||
ctx = ContractFlowContext(
|
||||
contract_url="http://contract-test.gzmpc.com:8080/",
|
||||
username="test002",
|
||||
password="1",
|
||||
supplier_name="合同节点联调供应商",
|
||||
supplier_name="重庆亿昊药业有限公司",
|
||||
order_date="2026-08-11",
|
||||
source_type="XYY",
|
||||
template_type="FORMAT",
|
||||
customer_code="DW0100046361",
|
||||
customer_name="重庆亿昊药业有限公司",
|
||||
goods_lines=[],
|
||||
)
|
||||
|
||||
run_contract_create_flow(browser, ctx)
|
||||
|
||||
assert browser.calls[:5] == [
|
||||
("open_url", "http://contract-test.gzmpc.com:8080/"),
|
||||
("fill_first", ('input[name="j_username"]', 'input[placeholder="用户名"]'), "test002"),
|
||||
("fill_first", ('input[name="j_password"]', 'input[placeholder="密码"]'), "1"),
|
||||
("click_first", ('input[name="btn_submit"]', 'a[href*="btn_submit"]', 'text=登录')),
|
||||
("wait_for_text", "合同", 20),
|
||||
]
|
||||
assert ("open_url", "http://contract-test.gzmpc.com:8080/") in browser.calls
|
||||
assert ("fill_first", ('input[name="j_username"]', 'input[placeholder="用户名"]'), "test002") in browser.calls
|
||||
assert ("fill_first", ('input[name="j_password"]', 'input[placeholder="密码"]'), "1") in browser.calls
|
||||
assert (
|
||||
"click_first",
|
||||
('a[class*="login_button"]', 'a:contains("登录")', 'input[name="btn_submit"]'),
|
||||
) in browser.calls
|
||||
assert not [call for call in browser.calls if call[0] == "screenshot"]
|
||||
|
||||
|
||||
def test_contract_flow_uses_order_date_and_uploads_yyc_merged_pdf():
|
||||
def _touch_pdf(path):
|
||||
path.write_bytes(b"%PDF-1.4\n1 0 obj<<>>endobj\ntrailer<<>>\n%%EOF\n")
|
||||
return str(path)
|
||||
|
||||
|
||||
def test_contract_flow_uses_order_date_and_uploads_yyc_merged_pdf(tmp_path):
|
||||
browser = FakeBrowser()
|
||||
browser.read_values["customer_name"] = "重庆亿昊药业有限公司"
|
||||
pdf_path = _touch_pdf(tmp_path / "contract-merged.pdf")
|
||||
ctx = ContractFlowContext(
|
||||
contract_url="http://contract-test.gzmpc.com:8080/",
|
||||
username="test002",
|
||||
password="1",
|
||||
supplier_name="合同节点联调供应商",
|
||||
supplier_name="重庆亿昊药业有限公司",
|
||||
order_date="2026-08-11",
|
||||
source_type="YYC",
|
||||
template_type="FORMAT",
|
||||
customer_code="DW0100046361",
|
||||
customer_name="重庆亿昊药业有限公司",
|
||||
goods_lines=[
|
||||
ContractGoodsLine(
|
||||
operation_code="ADD",
|
||||
goods_code="YP-001",
|
||||
goods_name="阿卡波糖片",
|
||||
approval_no="国药准字H00000001",
|
||||
spec="50mg*30片",
|
||||
goods_code="FQX0100118047",
|
||||
goods_name="盐酸托烷司琼注射液",
|
||||
approval_no="苏20160372",
|
||||
spec="5ml:5mg*5支",
|
||||
unit="盒",
|
||||
quantity=12,
|
||||
unit_price="50.00",
|
||||
)
|
||||
],
|
||||
merged_pdf_path="C:/rpa/jobs/JOB-1/contract-merged.pdf",
|
||||
merged_pdf_path=pdf_path,
|
||||
)
|
||||
|
||||
contract_no = run_contract_create_flow(browser, ctx)
|
||||
|
||||
assert contract_no == "HT-20260811-REAL-001"
|
||||
assert ("fill_first", ('label=合同签订日期', 'input[name="fdSignDate"]', 'input[placeholder*="合同签订"]'), "2026-08-11") in browser.calls
|
||||
assert ("fill_first", ('label=合同开始日期', 'input[name="fdStartDate"]', 'input[placeholder*="合同开始"]'), "2026-08-11") in browser.calls
|
||||
assert ("upload_file_first", ('input[type="file"]', 'label=合同证文上传'), "C:/rpa/jobs/JOB-1/contract-merged.pdf") in browser.calls
|
||||
assert contract_no == "001132026081734940"
|
||||
assert ("fill_first", ('input[name="fdSigndate"]',), "2026-08-11") in browser.calls
|
||||
assert ("fill_first", ('input[name="fdEffedate"]',), "2026-08-11") in browser.calls
|
||||
assert (
|
||||
"click_first",
|
||||
('div[title="选择商品"]', "div.lui-component.lui_widget_btn_txt", "text=选择商品"),
|
||||
) in browser.calls
|
||||
assert (
|
||||
"switch_to_frame",
|
||||
(
|
||||
'iframe[src*="EopBasedataMaterial"]',
|
||||
'iframe[src*="dialog_fdMaterial"]',
|
||||
'iframe[src*="common-template"]',
|
||||
),
|
||||
) in browser.calls
|
||||
assert ("fill_first", ("input.search-box-text", "input[placeholder*='关键字']"), "FQX0100118047") in browser.calls
|
||||
assert (
|
||||
"fill_first",
|
||||
('input[name="extendDataFormInfo.value(fd_detail1.0.fd_goodsqty)"]',),
|
||||
"12",
|
||||
) in browser.calls
|
||||
assert (
|
||||
"fill_first",
|
||||
('input[name="extendDataFormInfo.value(fd_detail1.0.fd_unitprice)"]',),
|
||||
"50.00",
|
||||
) in browser.calls
|
||||
assert any(
|
||||
call[0] == "upload_file_first" and call[-1] == pdf_path
|
||||
for call in browser.calls
|
||||
)
|
||||
|
||||
|
||||
def test_contract_flow_skips_pdf_upload_for_xyy():
|
||||
browser = FakeBrowser()
|
||||
browser.read_values["customer_name"] = "重庆亿昊药业有限公司"
|
||||
ctx = ContractFlowContext(
|
||||
contract_url="http://contract-test.gzmpc.com:8080/",
|
||||
username="test002",
|
||||
password="1",
|
||||
supplier_name="小药药供应商",
|
||||
supplier_name="重庆亿昊药业有限公司",
|
||||
order_date="2026-08-11",
|
||||
source_type="XYY",
|
||||
template_type="FORMAT",
|
||||
customer_code="DW0100046361",
|
||||
customer_name="重庆亿昊药业有限公司",
|
||||
goods_lines=[],
|
||||
merged_pdf_path="C:/should-not-upload.pdf",
|
||||
)
|
||||
|
|
@ -114,45 +273,46 @@ def test_contract_flow_skips_pdf_upload_for_xyy():
|
|||
assert not [call for call in browser.calls if call[0] == "upload_file_first"]
|
||||
|
||||
|
||||
def test_contract_flow_enters_real_draft_menu_and_recent_template():
|
||||
def test_contract_flow_enters_draft_form_by_template_url(tmp_path):
|
||||
from rpa_yisaiqi.contract_flow_steps import FORM_PAGE_URL
|
||||
|
||||
browser = FakeBrowser()
|
||||
browser.read_values["customer_name"] = "重庆亿昊药业有限公司"
|
||||
ctx = ContractFlowContext(
|
||||
contract_url="http://contract-test.gzmpc.com:8080/",
|
||||
username="test002",
|
||||
password="1",
|
||||
supplier_name="重庆亿昊药业",
|
||||
supplier_name="重庆亿昊药业有限公司",
|
||||
order_date="2026-08-11",
|
||||
source_type="YYC",
|
||||
template_type="NON_FORMAT",
|
||||
template_name="非格式--采购协议",
|
||||
customer_code="DW0100046361",
|
||||
customer_name="重庆亿昊药业有限公司",
|
||||
goods_lines=[],
|
||||
merged_pdf_path="C:/rpa/jobs/JOB-1/contract-merged.pdf",
|
||||
merged_pdf_path=_touch_pdf(tmp_path / "contract-merged.pdf"),
|
||||
)
|
||||
|
||||
run_contract_create_flow(browser, ctx)
|
||||
|
||||
assert ("click_first", ("text=合同管理", "aria=合同管理")) in browser.calls
|
||||
assert ("click_first", ("text=合同起草", "aria=合同起草")) in browser.calls
|
||||
assert ("wait_for_text", "最近使用模板", 20) in browser.calls
|
||||
assert (
|
||||
"click_first",
|
||||
("text=非格式--采购协议", "text=非格式--采购协议...", "title=非格式--采购协议"),
|
||||
) in browser.calls
|
||||
assert ("open_url", FORM_PAGE_URL) in browser.calls
|
||||
assert ("wait_for_text", "客户编码", 20) in browser.calls
|
||||
|
||||
|
||||
def test_contract_flow_fills_real_contract_draft_form_fields_from_screenshot():
|
||||
browser = FakeBrowser()
|
||||
browser.read_values["customer_name"] = "重庆亿昊药业有限公司"
|
||||
ctx = ContractFlowContext(
|
||||
contract_url="http://contract-test.gzmpc.com:8080/",
|
||||
username="test002",
|
||||
password="1",
|
||||
supplier_name="重庆亿昊药业",
|
||||
supplier_name="重庆亿昊药业有限公司",
|
||||
order_date="2026-08-11",
|
||||
source_type="XYY",
|
||||
template_type="NON_FORMAT",
|
||||
template_name="非格式--采购协议",
|
||||
customer_code="CUST-001",
|
||||
customer_name="重庆亿昊药业",
|
||||
customer_code="DW0100046361",
|
||||
customer_name="重庆亿昊药业有限公司",
|
||||
contract_category="采购协议",
|
||||
settlement_type="付款",
|
||||
contract_background="按订单自动起草采购协议",
|
||||
|
|
@ -172,37 +332,53 @@ def test_contract_flow_fills_real_contract_draft_form_fields_from_screenshot():
|
|||
|
||||
run_contract_create_flow(browser, ctx)
|
||||
|
||||
assert ("fill_first", ('label=客户编码', 'input[name="fdCustomerCode"]', 'input[placeholder*="客户编码"]'), "CUST-001") in browser.calls
|
||||
assert ("fill_first", ('label=客户名称', 'input[name="fdCustomerName"]', 'input[placeholder*="客户名称"]'), "重庆亿昊药业") in browser.calls
|
||||
assert ("select_first", ('label=合同类别', 'select[name="fdContractCategory"]'), "采购协议") in browser.calls
|
||||
assert ("select_first", ('label=结算类型', 'select[name="fdSettlementType"]'), "付款") in browser.calls
|
||||
assert ("fill_first", ('label=合同背景概述', 'textarea[name="fdContractBackground"]'), "按订单自动起草采购协议") in browser.calls
|
||||
assert ("click_first", ("label=主合同", "text=主合同")) in browser.calls
|
||||
assert ("select_first", ('label=合同金额', 'select[name="fdCurrency"]'), "人民币") in browser.calls
|
||||
assert ("fill_first", ('label=金额（元）', 'input[name="fdAmount"]'), "12000.50") in browser.calls
|
||||
assert ("fill_first", ('label=合同签订日期', 'input[name="fdSignDate"]', 'input[placeholder*="合同签订"]'), "2026-08-11") in browser.calls
|
||||
assert ("fill_first", ('label=合同开始日期', 'input[name="fdStartDate"]', 'input[placeholder*="合同开始"]'), "2026-08-11") in browser.calls
|
||||
assert ("fill_first", ('label=合同结束日期', 'input[name="fdEndDate"]', 'input[placeholder*="合同结束"]'), "2026-12-31") in browser.calls
|
||||
assert ("select_first", ('label=是否重大合同', 'select[name="fdIsMajorContract"]'), "否") in browser.calls
|
||||
assert ("select_first", ('label=是否预付款', 'select[name="fdIsPrepayment"]'), "否") in browser.calls
|
||||
assert ("fill_first", ('label=合同归属年份', 'input[name="fdBelongYear"]'), "2026") in browser.calls
|
||||
assert ("select_first", ('label=付款方式', 'select[name="fdPaymentMethod"]', 'select[name="paymentMethod"]'), "账期支付") in browser.calls
|
||||
assert ("fill_first", ('label=付款期限', 'input[name="fdPaymentDays"]', 'input[name="paymentDays"]'), "30") in browser.calls
|
||||
assert ("select_first", ('label=合同状态', 'select[name="fdContractStatus"]'), "正式合同") in browser.calls
|
||||
assert ("select_first", ('label=是否包含血液制品', 'select[name="fdContainsBloodProducts"]'), "否") in browser.calls
|
||||
assert (
|
||||
"select_xform",
|
||||
"extendDataFormInfo.value(fd_contract_cate_type)",
|
||||
"其他协议（采购）",
|
||||
) in browser.calls
|
||||
assert ("select_first", ('select#fdSettype',), "付款") in browser.calls
|
||||
assert (
|
||||
"fill_first",
|
||||
('textarea[name="extendDataFormInfo.value(fd_desc_info)"]',),
|
||||
"按订单自动起草采购协议",
|
||||
) in browser.calls
|
||||
assert ("select_xform", "extendDataFormInfo.value(fd_all_amount_type)", "合同标的金额") in browser.calls
|
||||
assert ("fill_first", ('input[name="fdAccount"]',), "12000.50") in browser.calls
|
||||
assert ("fill_first", ('input[name="fdSigndate"]',), "2026-08-11") in browser.calls
|
||||
assert ("fill_first", ('input[name="fdEffedate"]',), "2026-08-11") in browser.calls
|
||||
assert ("fill_first", ('input[name="fdEnddate"]',), "2026-12-31") in browser.calls
|
||||
assert ("select_xform", "extendDataFormInfo.value(fd_is_major_contract)", "否") in browser.calls
|
||||
assert ("select_xform", "extendDataFormInfo.value(fd_advance_amout)", "否") in browser.calls
|
||||
assert (
|
||||
"fill_first",
|
||||
('input[name="extendDataFormInfo.value(fd_contract_year_text)"]',),
|
||||
"2026",
|
||||
) in browser.calls
|
||||
assert ("select_xform", "extendDataFormInfo.value(fd_paytype)", "电汇") in browser.calls
|
||||
assert (
|
||||
"fill_first",
|
||||
('input[name="extendDataFormInfo.value(fd_pay_time_limit)"]',),
|
||||
"30",
|
||||
) in browser.calls
|
||||
assert ("select_xform", "extendDataFormInfo.value(fd_suconstatus)", "正式合同") in browser.calls
|
||||
assert ("select_xform", "extendDataFormInfo.value(fd_3d1ce3953f1162)", "否") in browser.calls
|
||||
|
||||
|
||||
def test_contract_flow_dry_run_skips_submit_but_still_reads_page_state():
|
||||
browser = FakeBrowser()
|
||||
browser.read_values["customer_name"] = "重庆亿昊药业有限公司"
|
||||
ctx = ContractFlowContext(
|
||||
contract_url="http://contract-test.gzmpc.com:8080/",
|
||||
username="test002",
|
||||
password="1",
|
||||
supplier_name="重庆亿昊药业",
|
||||
supplier_name="重庆亿昊药业有限公司",
|
||||
order_date="2026-08-11",
|
||||
source_type="XYY",
|
||||
template_type="NON_FORMAT",
|
||||
template_name="非格式--采购协议",
|
||||
customer_code="DW0100046361",
|
||||
customer_name="重庆亿昊药业有限公司",
|
||||
submit_mode="dry_run",
|
||||
goods_lines=[],
|
||||
)
|
||||
|
|
@ -210,6 +386,105 @@ def test_contract_flow_dry_run_skips_submit_but_still_reads_page_state():
|
|||
contract_no = run_contract_create_flow(browser, ctx)
|
||||
|
||||
assert not [call for call in browser.calls if call[0] == "click_first" and "text=提交" in call[1]]
|
||||
assert ("screenshot", "contract-dry-run-ready") in browser.calls
|
||||
assert not [call for call in browser.calls if call[0] == "screenshot"]
|
||||
assert contract_no == ""
|
||||
assert not [call for call in browser.calls if call[0] == "read_first"]
|
||||
|
||||
|
||||
def test_contract_flow_writes_text_accept_report_instead_of_requiring_screenshots(tmp_path, monkeypatch):
|
||||
from rpa_yisaiqi.contract_flow_steps import _ACCEPT, accept_log_path
|
||||
|
||||
monkeypatch.setenv("RPA_ACCEPT_LOG", str(tmp_path / "ACCEPT.txt"))
|
||||
browser = FakeBrowser()
|
||||
browser.read_values["customer_name"] = "重庆亿昊药业有限公司"
|
||||
ctx = ContractFlowContext(
|
||||
contract_url="http://contract-test.gzmpc.com:8080/",
|
||||
username="test002",
|
||||
password="1",
|
||||
supplier_name="重庆亿昊药业有限公司",
|
||||
order_date="2026-08-11",
|
||||
source_type="XYY",
|
||||
template_type="NON_FORMAT",
|
||||
customer_code="DW0100046361",
|
||||
customer_name="重庆亿昊药业有限公司",
|
||||
submit_mode="dry_run",
|
||||
goods_lines=[],
|
||||
)
|
||||
|
||||
run_contract_create_flow(browser, ctx)
|
||||
|
||||
text = accept_log_path("CONTRACT_CREATE").read_text(encoding="utf-8")
|
||||
assert "ACCEPT SUMMARY" in text
|
||||
assert "ACCEPT S03 ok" in text
|
||||
assert "ACCEPT S06 ok" in text
|
||||
assert "customerName=重庆亿昊药业有限公司" in text
|
||||
assert "ACCEPT S12 ok" in text
|
||||
assert all(line.ok for line in _ACCEPT.lines)
|
||||
|
||||
|
||||
def test_read_contract_number_falls_back_to_my_contracts_list():
|
||||
from rpa_yisaiqi.contract_flow_steps import read_contract_number
|
||||
|
||||
browser = FakeBrowser()
|
||||
browser.read_values.pop("contract_number", None)
|
||||
ctx = ContractFlowContext(
|
||||
contract_url="http://contract-test.gzmpc.com:8080/",
|
||||
username="test002",
|
||||
password="1",
|
||||
supplier_name="重庆亿昊药业有限公司",
|
||||
order_date="2026-08-11",
|
||||
source_type="XYY",
|
||||
customer_code="DW0100046361",
|
||||
customer_name="重庆亿昊药业有限公司",
|
||||
contract_title="test1",
|
||||
goods_lines=[],
|
||||
)
|
||||
|
||||
contract_no = read_contract_number(browser, ctx)
|
||||
|
||||
assert contract_no == "001132026081734940"
|
||||
assert ("open_url", "http://contract-test.gzmpc.com:8080/") in browser.calls
|
||||
assert ("click_first", ("text=我的合同",)) in browser.calls
|
||||
assert ("click_first", ("text=我起草的",)) in browser.calls
|
||||
assert ("click_first", ("text=申请时间",)) in browser.calls
|
||||
|
||||
|
||||
def test_read_contract_number_goes_to_my_list_even_without_ctx_arg():
|
||||
from rpa_yisaiqi import _flow_state as state
|
||||
from rpa_yisaiqi.contract_flow_steps import read_contract_number
|
||||
|
||||
browser = FakeBrowser()
|
||||
browser.read_values.pop("contract_number", None)
|
||||
old_ctx = state.ctx
|
||||
state.ctx = ContractFlowContext(
|
||||
contract_url="http://contract-test.gzmpc.com:8080/",
|
||||
username="test002",
|
||||
password="1",
|
||||
supplier_name="重庆亿昊药业有限公司",
|
||||
order_date="2026-08-11",
|
||||
source_type="XYY",
|
||||
customer_code="DW0100046361",
|
||||
customer_name="重庆亿昊药业有限公司",
|
||||
contract_title="test1",
|
||||
goods_lines=[],
|
||||
)
|
||||
try:
|
||||
contract_no = read_contract_number(browser)
|
||||
finally:
|
||||
state.ctx = old_ctx
|
||||
|
||||
assert contract_no == "001132026081734940"
|
||||
assert ("click_first", ("text=我的合同",)) in browser.calls
|
||||
assert ("click_first", ("text=申请时间",)) in browser.calls
|
||||
|
||||
|
||||
def test_accept_check_fails_when_page_value_mismatches():
|
||||
from rpa_yisaiqi.contract_flow_steps import AcceptanceReport
|
||||
from rpa_yisaiqi.runtime.errors import NodeError
|
||||
|
||||
report = AcceptanceReport()
|
||||
with pytest.raises(NodeError, match="ACCEPT S06 FAIL"):
|
||||
report.check(
|
||||
"S06",
|
||||
"签约方",
|
||||
[("customerName", "其他公司", "重庆亿昊药业有限公司")],
|
||||
)
|
||||
|
|
|
|||
Loading…
Reference in New Issue