AutoRpa/rpa_rebate/tests/test_finder.py

from accept import accept_log_path
from job import run_batch
from persist import KIND_OK, USER_NAME, AgreementOutcome
from rebate_finder import RebateFinder
from rebate_flow_steps import (
    SENTINEL_CONTRACT_NO,
    RebateFlowContext,
    classify_search_result,
    run_find_flow,
    search_contract,
)
from repo import MemoryRebateRepo


class FakeBrowser:
    def __init__(self, *, table_text="", body_text="", login_body=""):
        self.table_text = table_text
        self.body_text = body_text
        self.login_body = login_body or body_text
        self.opened: list[str] = []
        self.filled: 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
        return self.table_text

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

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

    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:
        return None

    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 test_table_row_with_contract_no_is_found():
    assert classify_search_result("HT1\n重庆亿昊药业有限公司", "HT1", from_table=True) is True


def test_empty_table_is_not_found():
    assert classify_search_result("暂无数据", "HT1", from_table=True) is False


def test_search_box_echo_is_not_found():
    browser = FakeBrowser(
        table_text="暂无数据",
        body_text="合同编号 HT1 查询 暂无数据",
    )
    assert search_contract(browser, "HT1") is False
    assert browser.filled[-1][1] == "HT1"
    assert browser.clicked


def test_search_table_hit_is_found():
    browser = FakeBrowser(table_text="合同编号 供应商\nHT1 重庆亿昊药业有限公司")
    assert search_contract(browser, "HT1") is True


def test_search_one_row_without_visible_contract_no_is_found():
    browser = FakeBrowser(
        table_text="协议总单ID 采购协议号\nCG-743799-20261010-00002",
        body_text="采购协议管理 共 1 条",
    )
    assert search_contract(browser, "035012026101035440") is True


def test_page_finder_found_writes_agreement_like_stub():
    repo = MemoryRebateRepo(
        table4=[
            {
                "fid": 1,
                "oms_id": 2026100501,
                "action_code": 1,
                "contract_status": 1,
                "contract_fail_reason": None,
                "contract_no": "HT1",
                "agreement_status": None,
                "agreement_fail_reason": None,
                "agreement_id": None,
                "agreement_dtl_id": None,
                "org_code": "901",
                "push_status": 0,
                "process_status": 0,
                "create_user_name": USER_NAME,
                "update_user_name": USER_NAME,
            }
        ]
    )
    browser = FakeBrowser(
        table_text="HT1 重庆亿昊",
        body_text="采购协议管理 合同编号 查询",
        login_body="采购协议管理",
    )
    finder = RebateFinder(
        browser,
        rebate_url="http://example",
        username="4795",
        password="secret",
    )

    def runner(group):
        assert finder(group.contract_no) is True
        return AgreementOutcome(KIND_OK, "AGR1", "AGRD1", None)

    summary = run_batch(repo, runner=runner)
    assert summary.ok == 1
    assert repo.table4[1]["contract_status"] == 2
    assert repo.table4[1]["process_status"] == 1
    assert repo.table4[1]["update_user_name"] == USER_NAME
    agreements = [row for row in repo.table4.values() if row["action_code"] == 2]
    assert len(agreements) == 1
    assert len(browser.opened) == 1


def test_find_flow_writes_accept_for_missing_number(tmp_path, monkeypatch):
    monkeypatch.delenv("RPA_ACCEPT_LOG", raising=False)
    monkeypatch.setattr("accept.SCREENSHOT_ROOT", tmp_path)
    browser = FakeBrowser(
        table_text="暂无数据",
        body_text="采购协议管理 合同编号 查询",
    )
    ctx = RebateFlowContext(rebate_url="http://example", username="u", password="p")
    found = run_find_flow(browser, ctx, SENTINEL_CONTRACT_NO, expect_found=False)
    assert found is False
    text = accept_log_path("REBATE_FIND").read_text(encoding="utf-8")
    assert "ACCEPT SUMMARY REBATE_FIND 4/4 ok" in text
    assert "ACCEPT S05 ok 查号" in text
    assert f"contractNo={SENTINEL_CONTRACT_NO}" in text
    assert "found=no" in text