gyxx-flow/tests/test_supply_acceptance_isolation.py

from __future__ import annotations

import importlib
import json
from pathlib import Path

import pytest

from gyxx_flow.adapters import WORKFLOW_ACCEPTANCE_RECIPIENT_OPEN_ID
from gyxx_flow.notification_routing import ANALYZER_ACCEPTANCE_RECIPIENT_OPEN_ID


@pytest.fixture
def acceptance_environment(
    tmp_path: Path,
    monkeypatch: pytest.MonkeyPatch,
) -> Path:
    evidence = tmp_path / "evidence.jsonl"
    monkeypatch.setenv("GYXX_WORKFLOW_ACCEPTANCE", "1")
    monkeypatch.setenv("GYXX_FEISHU_TABLE_WRITE_DISABLED", "1")
    monkeypatch.setenv("GYXX_COOKIE_INVALID_SKIP", "1")
    monkeypatch.setenv(
        "GYXX_NOTIFICATION_RECIPIENT_OPEN_ID",
        WORKFLOW_ACCEPTANCE_RECIPIENT_OPEN_ID,
    )
    monkeypatch.setenv("GYXX_ACCEPTANCE_EVIDENCE_FILE", str(evidence))
    return evidence


def test_replenishment_bitable_run_is_physically_skipped(
    tmp_path: Path,
    acceptance_environment: Path,
    monkeypatch: pytest.MonkeyPatch,
) -> None:
    module = importlib.import_module(
        "gyxx_flow.modules.supply_chain.orchestrator.scripts."
        "insert_replenishment_bitable"
    )
    pending = tmp_path / "pending_insert.json"
    pending.write_text(
        json.dumps(
            {
                "pending_count": 2,
                "records": [
                    {"商品编号": "SKU1"},
                    {"商品编号": "SKU2"},
                ],
            },
            ensure_ascii=False,
        ),
        encoding="utf-8",
    )
    monkeypatch.setattr(
        module,
        "_search_record_id_by_sku",
        lambda *_args, **_kwargs: pytest.fail("must not access Feishu"),
    )

    result = module.run(pending)

    assert result == {
        "inserted": 0,
        "updated": 0,
        "skipped": 2,
        "record_ids": [],
        "errors": [],
        "skipped_by_policy": True,
    }
    assert "feishu_write_skipped" in acceptance_environment.read_text(
        encoding="utf-8"
    )


def test_supply_card_sender_forces_only_wang_yunlong(
    acceptance_environment: Path,
    monkeypatch: pytest.MonkeyPatch,
    tmp_path: Path,
) -> None:
    module = importlib.import_module(
        "gyxx_flow.modules.supply_chain.orchestrator.scripts."
        "send_card_notification"
    )
    monkeypatch.setattr(module, "STATE_ROOT", tmp_path)
    run_id = "acceptance-run"
    snapshot_path = module._notification_snapshot_path("replenishment", run_id)
    snapshot_path.parent.mkdir(parents=True)
    snapshot_path.write_text(
        json.dumps(
            {
                "schema_version": 1,
                "workflow": "replenishment",
                "run_id": run_id,
                "enabled": True,
                "recipients": [ANALYZER_ACCEPTANCE_RECIPIENT_OPEN_ID],
                "acceptance": True,
            }
        ),
        encoding="utf-8",
    )
    sent: list[str] = []
    monkeypatch.setattr(
        module,
        "_send_card",
        lambda open_id, _card, _key: sent.append(open_id) or "om_acceptance1",
    )
    monkeypatch.setattr(
        module.pg_writer,
        "mark_notification_sent",
        lambda *_args, **_kwargs: None,
    )
    monkeypatch.setattr(
        module.pg_writer,
        "notification_already_sent",
        lambda *_args, **_kwargs: True,
    )

    with pytest.raises(RuntimeError, match="snapshot mismatch"):
        module.run(
            ["ou_other123"],
            {"elements": []},
            workflow="replenishment",
            run_id=run_id,
            idempotency_key="replenishment:acceptance-run",
        )

    result = module.run(
        [ANALYZER_ACCEPTANCE_RECIPIENT_OPEN_ID],
        {"elements": []},
        workflow="replenishment",
        run_id=run_id,
        idempotency_key="replenishment:acceptance-run",
    )

    assert sent == [ANALYZER_ACCEPTANCE_RECIPIENT_OPEN_ID]
    assert result == {
        "message_ids": ["om_acceptance1"],
        "recipient_message_ids": {
            ANALYZER_ACCEPTANCE_RECIPIENT_OPEN_ID: "om_acceptance1",
        },
        "errors": [],
    }


def test_supply_config_uses_the_two_local_hermes_api_servers(
    acceptance_environment: Path,
    monkeypatch: pytest.MonkeyPatch,
) -> None:
    monkeypatch.setenv(
        "COLLECTOR_API_SERVER_URL",
        "http://127.0.0.1:8643/v1/chat/completions",
    )
    monkeypatch.setenv(
        "ANALYZER_API_SERVER_URL",
        "http://127.0.0.1:8642/v1/chat/completions",
    )
    module = importlib.import_module(
        "gyxx_flow.modules.supply_chain.orchestrator.config"
    )
    module = importlib.reload(module)

    assert module.HERMES_GATEWAYS["collector"]["url"] == "http://127.0.0.1:8643/v1"
    assert module.HERMES_GATEWAYS["collector"]["model"] == "data-collector"
    assert module.HERMES_GATEWAYS["analyzer"]["url"] == "http://127.0.0.1:8642/v1"
    assert module.HERMES_GATEWAYS["analyzer"]["model"] == "data-analyzer"
    for targets in module.FEISHU_CONFIG["notification_targets"].values():
        assert targets == [WORKFLOW_ACCEPTANCE_RECIPIENT_OPEN_ID]


def test_purchase_order_update_is_blocked_before_erp_mutation(
    acceptance_environment: Path,
) -> None:
    module = importlib.import_module(
        "gyxx_flow.modules.supply_chain.orchestrator.mcp_workflow"
    )
    module = importlib.reload(module)

    result = module.run_mcp_workflow(
        "purchase-order-update",
        "acceptance-purchase-order-update",
    )

    assert result["status"] == "blocked_business_mutation"
    assert "approved test task/SKU fixture" in result["error"]
    assert "business_mutation_blocked" in acceptance_environment.read_text(
        encoding="utf-8"
    )