AutoRpa/rpa_rebate/tests/test_submit.py

from accept import accept_log_path
from apa_browser_adapter import chrome_debug_port, chrome_session_flow, resolve_chrome_profile_dir
from rebate_flow_steps import LOGIN, RebateFlowContext, _clean_submit_id, run_submit_flow


class FakeBrowser:
    def __init__(self, *, table_text="", body_text="", login_body="", agreement_id="", agreement_dtl_id=""):
        self.table_text = table_text
        self.body_text = body_text
        self.login_body = login_body or body_text
        self.agreement_id = agreement_id
        self.agreement_dtl_id = agreement_dtl_id
        self.opened: list[str] = []
        self.filled: list[tuple[list[str], str]] = []
        self.selected: list[tuple[list[str], str]] = []
        self.clicked: list[list[str]] = []
        self.script_count = 0

    def open_url(self, url: str) -> None:
        self.opened.append(url)

    def fill_first(self, selectors: list[str], value: str) -> None:
        self.filled.append((list(selectors), value))

    def click_first(self, selectors: list[str]) -> None:
        self.clicked.append(list(selectors))

    def click_native(self, selectors: list[str]) -> None:
        self.click_first(selectors)

    def wait_for_text(self, text: str, timeout_seconds: int = 15) -> None:
        haystack = f"{self.body_text}\n{self.table_text}\n{self.login_body}"
        if text not in haystack:
            raise TimeoutError(text)

    def wait_for_element(self, selectors: list[str], timeout_seconds: int = 15) -> None:
        return None

    def element_exists(self, selectors: list[str]) -> bool:
        return True

    def read_body_text(self) -> str:
        return self.body_text

    def execute_script(self, script: str, *args):
        self.script_count += 1
        text = script or ""
        if "SELECT_HEADER_ROW" in text or "OPEN_HEADER_ROW" in text:
            return {"ok": True, "via": "row"}
        if "READ_AGREEMENT_NOS" in text:
            return {"header": self.agreement_id, "detail": self.agreement_dtl_id}
        if "READ_REBATE_ORDER_NO" in text:
            return self.agreement_id
        if "CONFIRM_SUBMIT" in text:
            return {"ok": False, "asked": False}
        if "table-toolbar-btn__text" in text:
            title = str(args[0] if args else "")
            self.clicked.append([title])
            return {"ok": True}
        return self.table_text

    def screenshot(self, name: str) -> str:
        return name

    def read_first(self, selectors: list[str], value_name: str = "") -> str:
        return self.agreement_id

    def switch_to_frame(self, selectors: list[str]) -> None:
        return None

    def switch_to_default_content(self) -> None:
        return None

    def select_first(self, selectors: list[str], value: str) -> None:
        self.selected.append((list(selectors), value))

    def select_xform(self, name: str, value: str) -> None:
        return None

    def upload_file_first(self, selectors: list[str], file_path: str) -> None:
        return None


def _ctx() -> RebateFlowContext:
    return RebateFlowContext(rebate_url="http://example", username="u", password="p")


def test_submit_flow_writes_accept_with_order_nos(tmp_path, monkeypatch):
    monkeypatch.delenv("RPA_ACCEPT_LOG", raising=False)
    monkeypatch.setattr("accept.SCREENSHOT_ROOT", tmp_path)
    browser = FakeBrowser(
        table_text="HT1 重庆亿昊",
        body_text="采购协议管理 合同编号 查询 送审 协议总单ID 71342020260528 协议细单ID 71342020260529",
        login_body="采购协议管理",
        agreement_id="71342020260528",
        agreement_dtl_id="71342020260529",
    )
    result = run_submit_flow(browser, _ctx(), "HT1")
    assert result.submitted is True
    assert result.agreement_id == "71342020260528"
    assert result.agreement_dtl_id == "71342020260529"
    assert ["送审"] in browser.clicked
    text = accept_log_path("REBATE_SUBMIT").read_text(encoding="utf-8")
    assert "ACCEPT SUMMARY REBATE_SUBMIT 7/7 ok" in text
    assert "ACCEPT S07 ok 点送审" in text
    assert "submitted=yes" in text
    assert "ACCEPT S08 ok 读号" in text
    assert "rebateOrderNo=71342020260528" in text
    assert "agreementDtlId=71342020260529" in text


def test_submit_reuses_logged_in_session(tmp_path, monkeypatch):
    monkeypatch.delenv("RPA_ACCEPT_LOG", raising=False)
    monkeypatch.setattr("accept.SCREENSHOT_ROOT", tmp_path)
    browser = FakeBrowser(
        table_text="HT1 重庆亿昊",
        body_text="采购协议管理 待审批 合同编号 查询 送审",
        login_body="采购协议管理",
        agreement_id="AGR1",
        agreement_dtl_id="AGRD1",
    )
    run_submit_flow(browser, _ctx(), "HT1")
    login_fills = [value for selectors, value in browser.filled if selectors == LOGIN["user"]]
    assert login_fills == []
    assert "u" not in [value for _, value in browser.filled]


def test_submit_chrome_session_reuses_find():
    assert chrome_session_flow("REBATE_SUBMIT") == "REBATE_FIND"
    assert resolve_chrome_profile_dir("REBATE_SUBMIT") == resolve_chrome_profile_dir("REBATE_FIND")
    assert chrome_debug_port("REBATE_SUBMIT") == chrome_debug_port("REBATE_FIND")


def test_submit_scan_rejects_loading_placeholder():
    assert _clean_submit_id("查询加载中...设置查询条件") == ""
    assert _clean_submit_id("协议细单ID") == ""
    assert _clean_submit_id("71342020260528") == "71342020260528"
    assert _clean_submit_id("71342020260528", contract_no="71342020260528") == ""