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="重庆亿昊药业有限公司", 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 ("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 _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="重庆亿昊药业有限公司", order_date="2026-08-11", source_type="YYC", template_type="FORMAT", customer_code="DW0100046361", customer_name="重庆亿昊药业有限公司", goods_lines=[ ContractGoodsLine( operation_code="ADD", goods_code="FQX0100118047", goods_name="盐酸托烷司琼注射液", approval_no="苏20160372", spec="5ml:5mg*5支", unit="盒", quantity=12, unit_price="50.00", ) ], merged_pdf_path=pdf_path, ) contract_no = run_contract_create_flow(browser, ctx) 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="重庆亿昊药业有限公司", 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", ) run_contract_create_flow(browser, ctx) assert not [call for call in browser.calls if call[0] == "upload_file_first"] 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="重庆亿昊药业有限公司", order_date="2026-08-11", source_type="YYC", template_type="NON_FORMAT", template_name="非格式--采购协议", customer_code="DW0100046361", customer_name="重庆亿昊药业有限公司", goods_lines=[], merged_pdf_path=_touch_pdf(tmp_path / "contract-merged.pdf"), ) run_contract_create_flow(browser, ctx) 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="重庆亿昊药业有限公司", order_date="2026-08-11", source_type="XYY", template_type="NON_FORMAT", template_name="非格式--采购协议", customer_code="DW0100046361", customer_name="重庆亿昊药业有限公司", contract_category="采购协议", settlement_type="付款", contract_background="按订单自动起草采购协议", contract_nature="主合同", amount_currency="人民币", amount="12000.50", contract_end_date="2026-12-31", is_major_contract="否", is_prepayment="否", belong_year="2026", payment_method="账期支付", payment_days="30", contract_status="正式合同", contains_blood_products="否", goods_lines=[], ) run_contract_create_flow(browser, ctx) 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="重庆亿昊药业有限公司", 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=[], ) 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 not [call for call in browser.calls if call[0] == "screenshot"] assert contract_no == "" 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", "其他公司", "重庆亿昊药业有限公司")], )