gyxx-flow/tests/test_supply_replenishment_arrival_sync.py

from __future__ import annotations

import json
from contextlib import contextmanager
from datetime import date, datetime
from pathlib import Path
from types import SimpleNamespace

import pytest

from gyxx_flow.catalog import WorkflowCatalog
from gyxx_flow.modules.supply_chain.orchestrator.arrival_sync_queue import (
    ArrivalSyncQueue,
    ArrivalSyncTask,
)
from gyxx_flow.modules.supply_chain.orchestrator.scripts import (
    PurchaseOrderUpdate,
    sync_replenishment_arrivals,
)
from gyxx_flow.modules.supply_chain.orchestrator.scripts.PurchaseOrderUpdate import (
    ERPLoginManager,
)
from gyxx_flow.modules.supply_chain.orchestrator.scripts.sync_replenishment_arrivals import (
    arrival_date_text,
)

PROJECT_ROOT = Path(__file__).resolve().parents[1]


class _Cursor:
    def __init__(self, connection: "_Connection") -> None:
        self.connection = connection

    def __enter__(self) -> "_Cursor":
        return self

    def __exit__(self, *_args: object) -> None:
        return None

    def execute(self, sql: str, params: tuple[object, ...]) -> None:
        self.connection.calls.append((sql, params))

    def fetchall(self) -> list[dict[str, object]]:
        return self.connection.claim_rows

    def fetchone(self) -> dict[str, object] | None:
        return self.connection.return_row


class _Connection:
    def __init__(self, *, claim_rows: list[dict[str, object]] | None = None) -> None:
        self.claim_rows = claim_rows or []
        self.return_row: dict[str, object] | None = {"id": 1}
        self.calls: list[tuple[str, tuple[object, ...]]] = []

    def __enter__(self) -> "_Connection":
        return self

    def __exit__(self, *_args: object) -> None:
        return None

    @contextmanager
    def cursor(self, **_kwargs: object):
        yield _Cursor(self)


def test_arrival_date_text_uses_date_part_and_supports_clear() -> None:
    assert arrival_date_text(datetime(2026, 8, 20, 15, 30)) == "2026-08-20"
    assert arrival_date_text(date(2026, 8, 21)) == "2026-08-21"
    assert arrival_date_text("2026-08-22T00:00:00") == "2026-08-22"
    assert arrival_date_text(None) == ""


def test_arrival_date_text_rejects_invalid_values() -> None:
    with pytest.raises(ValueError, match="格式无效"):
        arrival_date_text("not-a-date")


def test_detail_row_lookup_uses_loaded_sku_row_not_ambiguous_tab_name() -> None:
    class _Cell:
        def __init__(self, text: str) -> None:
            self.text = text

        def inner_text(self) -> str:
            return self.text

    class _Cells:
        def __init__(self, values: list[str]) -> None:
            self.values = values

        def count(self) -> int:
            return len(self.values)

        def nth(self, index: int) -> _Cell:
            return _Cell(self.values[index])

    class _Row:
        def __init__(self, values: list[str]) -> None:
            self.values = values

        def locator(self, selector: str) -> _Cells:
            assert selector == "._jt_cell"
            return _Cells(self.values)

    class _Rows:
        def __init__(self, rows: list[_Row]) -> None:
            self.rows = rows

        def count(self) -> int:
            return len(self.rows)

        def nth(self, index: int) -> _Row:
            return self.rows[index]

    class _Frame:
        def __init__(self, rows: list[_Row]) -> None:
            self.rows = rows

        def locator(self, selector: str) -> _Rows:
            assert selector == "div._jt_row._jt_rh, div._jt_row"
            return _Rows(self.rows)

    irrelevant_tab0 = _Frame([_Row(["973798", "采购单列表"])] )
    purchase_detail = _Frame([_Row(["10504002", "星迹2", "时野绿中号"])] )
    page = SimpleNamespace(frames=[irrelevant_tab0, purchase_detail])

    frame, row = PurchaseOrderUpdate.PurchaseManager._find_target_detail_row(
        page, "10504002"
    )

    assert frame is purchase_detail
    assert row is not None


def test_detail_filter_renders_the_target_sku_in_a_virtualized_grid() -> None:
    calls: list[tuple[str, str]] = []

    class _SearchBox:
        first: "_SearchBox"

        def __init__(self) -> None:
            self.first = self

        @staticmethod
        def count() -> int:
            return 1

        def fill(self, value: str) -> None:
            calls.append(("fill", value))

        def press(self, value: str) -> None:
            calls.append(("press", value))

    class _Frame:
        @staticmethod
        def locator(selector: str) -> _SearchBox:
            assert selector == "#sku_search"
            return _SearchBox()

    page = SimpleNamespace(frames=[_Frame()])

    assert PurchaseOrderUpdate.PurchaseManager._filter_purchase_detail_by_sku(
        page, "10504002"
    )
    assert calls == [("fill", "10504002"), ("press", "Enter")]


def test_retryable_item_failures_do_not_fail_the_worker() -> None:
    result = {
        "errors": ["SKU 10465007: 搜索商品编码失败"],
        "retryable_failures": ["SKU 10465007: 搜索商品编码失败"],
        "execution_errors": [],
    }

    assert sync_replenishment_arrivals._worker_exit_code(result) == 0


def test_execution_errors_still_fail_the_worker() -> None:
    result = {
        "errors": ["队列 PostgreSQL 连接失败"],
        "retryable_failures": [],
        "execution_errors": ["队列 PostgreSQL 连接失败"],
    }

    assert sync_replenishment_arrivals._worker_exit_code(result) == 1


def test_run_records_item_failure_for_retry_without_failing_worker(
    tmp_path: Path,
    monkeypatch: pytest.MonkeyPatch,
) -> None:
    config = tmp_path / "config.json"
    config.write_text("{}", encoding="utf-8")
    task = ArrivalSyncTask(
        id=8,
        tenant_id=1,
        sku_code="SKU-8",
        expected_arrival_time=datetime(2026, 8, 20, 9, 0),
    )

    class _Queue:
        def __init__(self) -> None:
            self.failed: list[tuple[ArrivalSyncTask, str]] = []

        def claim(self, **_kwargs: object) -> list[ArrivalSyncTask]:
            return [task]

        def mark_failed(self, claimed_task: ArrivalSyncTask, error: str) -> bool:
            self.failed.append((claimed_task, error))
            return True

    class _Manager:
        _acceptance_cookie_skipped = False

        def __init__(self, *_args: object, **_kwargs: object) -> None:
            return None

        def login_with_browser(self) -> bool:
            return True

        def open_purchase_page(self) -> bool:
            return True

        def search_sku(self, _sku_code: str) -> bool:
            return False

        def close_browser(self) -> None:
            return None

    monkeypatch.setattr(sync_replenishment_arrivals, "ArrivalSyncQueue", _Queue)
    monkeypatch.setattr(sync_replenishment_arrivals, "CONFIG_PATH", config)
    monkeypatch.setattr(
        sync_replenishment_arrivals,
        "WORK_ROOT",
        tmp_path / "work",
    )
    monkeypatch.setattr(
        sync_replenishment_arrivals,
        "current_acceptance_policy",
        lambda: SimpleNamespace(enabled=False),
    )
    monkeypatch.setattr(PurchaseOrderUpdate, "PurchaseManager", _Manager)
    monkeypatch.setenv("GYXX_RUN_ID", "retryable-item-failure")

    assert sync_replenishment_arrivals.run(cdp_port=22139) == 0

    result_path = (
        tmp_path
        / "work"
        / "arrival-sync"
        / "run_id=retryable-item-failure"
        / "result.json"
    )
    result = json.loads(result_path.read_text(encoding="utf-8"))
    assert result["failed"] == 1
    assert result["retryable_failures"] == ["SKU SKU-8: 搜索商品编码失败"]
    assert result["execution_errors"] == []


def test_arrival_sync_workflow_uses_daily_python_browser_entry() -> None:
    catalog = WorkflowCatalog.load(PROJECT_ROOT / "config")
    workflow = next(
        item
        for item in catalog.scheduled_workflows()
        if item.workflow_id == "supply.replenishment_arrival_sync.daily"
    )

    assert workflow.steps[0].entry == (
        "orchestrator/scripts/collect_replenishment_arrival_sync.py"
    )
    assert workflow.steps[0].replay_policy == "idempotent"
    assert catalog.schedule_for(workflow.workflow_id).at == "09:00"


def test_queue_claims_rows_with_skip_locked_and_maps_tasks() -> None:
    connection = _Connection(
        claim_rows=[
            {
                "id": 8,
                "tenant_id": 1,
                "sku_code": " SKU-8 ",
                "expected_arrival_time": datetime(2026, 8, 20, 9, 0),
            }
        ]
    )
    queue = ArrivalSyncQueue(
        {"GYXX_ARRIVAL_SYNC_OPERATOR": "worker"},
        connect_factory=lambda: connection,
    )

    tasks = queue.claim(tenant_id=1, stale_after_seconds=60)

    assert tasks == [
        ArrivalSyncTask(
            id=8,
            tenant_id=1,
            sku_code="SKU-8",
            expected_arrival_time=datetime(2026, 8, 20, 9, 0),
        )
    ]
    assert "FOR UPDATE SKIP LOCKED" in connection.calls[0][0]
    assert "LIMIT" not in connection.calls[0][0]
    assert "deleted = 0" in connection.calls[0][0]
    assert "deleted = FALSE" not in connection.calls[0][0]
    assert connection.calls[0][1] == (1, 60, "worker")


def test_queue_completion_is_fenced_by_the_claimed_target_timestamp() -> None:
    connection = _Connection()
    queue = ArrivalSyncQueue(connect_factory=lambda: connection)
    task = ArrivalSyncTask(
        id=8,
        tenant_id=1,
        sku_code="SKU-8",
        expected_arrival_time=None,
    )

    assert queue.mark_success(task) is True
    success_sql, success_params = connection.calls[-1]
    assert "IS NOT DISTINCT FROM" in success_sql
    assert success_params == ("gyxx-flow", 8, 1, None)

    connection.return_row = None
    assert queue.mark_failed(task, "  浏览器失败\n原因 ") is False
    failed_sql, failed_params = connection.calls[-1]
    assert "IS NOT DISTINCT FROM" in failed_sql
    assert failed_params == ("浏览器失败 原因", "gyxx-flow", 8, 1, None)


def test_cdp_session_is_checked_before_optional_password_validation(
    tmp_path: Path,
    monkeypatch: pytest.MonkeyPatch,
) -> None:
    config = tmp_path / "config.json"
    config.write_text(
        '{"login": {"url": "https://login.example.test", '
        '"username": "erp-user"}}',
        encoding="utf-8",
    )
    monkeypatch.delenv("GYXX_SUPPLY_ERP_PASSWORD", raising=False)
    monkeypatch.delenv("GYXX_WORKFLOW_ACCEPTANCE", raising=False)

    class _Page:
        class _Context:
            @staticmethod
            def cookies() -> list[dict[str, str]]:
                return [{"name": "token", "value": "session"}]

        context = _Context()

    manager = ERPLoginManager(str(config), cdp_port=22139)
    page = _Page()
    monkeypatch.setattr(manager, "_prepare_login_page", lambda *_args, **_kwargs: page)
    monkeypatch.setattr(
        manager,
        "_page_has_purchase_access",
        lambda *_args, **_kwargs: True,
    )

    assert manager.login_with_browser() is True