shanxia2.0/backend/tests/test_sync_plan_contract_no_snapshots.py

"""CRM 合同改号后，运输计划合同号快照应同步。"""
from datetime import date
from unittest.mock import MagicMock, patch

import pytest
from sqlalchemy import create_engine
from sqlalchemy.orm import Session

from app.models import pl_models as m


@pytest.fixture
def db():
    engine = create_engine("sqlite://")
    m.PlPlan.__table__.create(engine, checkfirst=True)
    m.PlPlanContract.__table__.create(engine, checkfirst=True)
    with Session(engine, autoflush=False) as session:
        yield session
    engine.dispose()


def _seed(db: Session):
    plan = m.PlPlan(
        id="plan-1",
        plan_no="PO20261012005",
        create_date=date(2026, 10, 12),
        belongs_company="shanxia",
        plan_type="direct",
        contract_id="contract-1",
        contract_no="SX/X-JT20261009002",
        tenant_no="t1",
    )
    other_plan = m.PlPlan(
        id="plan-2",
        plan_no="PO20261012099",
        create_date=date(2026, 10, 12),
        belongs_company="shanxia",
        plan_type="direct",
        contract_id="contract-other",
        contract_no="OTHER-001",
        tenant_no="t1",
    )
    link = m.PlPlanContract(
        id="link-1",
        plan_id="plan-1",
        side="delivery",
        stop_id="stop-1",
        contract_id="contract-1",
        contract_no="SX/X-JT20261009002",
        tenant_no="t1",
    )
    deleted = m.PlPlanContract(
        id="link-del",
        plan_id="plan-1",
        side="delivery",
        contract_id="contract-1",
        contract_no="SX/X-JT20261009002",
        tenant_no="t1",
        del_flag=1,
    )
    other_tenant = m.PlPlanContract(
        id="link-tenant",
        plan_id="plan-x",
        side="delivery",
        contract_id="contract-1",
        contract_no="SX/X-JT20261009002",
        tenant_no="other",
    )
    db.add_all([plan, other_plan, link, deleted, other_tenant])
    db.flush()
    return plan, link


def test_sync_refreshes_plan_link_and_legacy_header(db):
    from app.services.pl_plan_contract_no_sync import sync_plan_contract_no_snapshots

    plan, link = _seed(db)
    changed = sync_plan_contract_no_snapshots(
        db,
        tenant_no="t1",
        contract_id="contract-1",
        contract_no="P0-2026-10-0016",
        user_id="u1",
    )
    db.flush()
    db.expire_all()

    assert changed >= 2
    assert db.get(m.PlPlanContract, "link-1").contract_no == "P0-2026-10-0016"
    assert db.get(m.PlPlan, "plan-1").contract_no == "P0-2026-10-0016"
    assert db.get(m.PlPlanContract, "link-1").update_by == "u1"
    assert db.get(m.PlPlan, "plan-2").contract_no == "OTHER-001"
    assert db.get(m.PlPlanContract, "link-del").contract_no == "SX/X-JT20261009002"
    assert db.get(m.PlPlanContract, "link-tenant").contract_no == "SX/X-JT20261009002"


def test_sync_noop_when_already_current(db):
    from app.services.pl_plan_contract_no_sync import sync_plan_contract_no_snapshots

    _seed(db)
    sync_plan_contract_no_snapshots(
        db,
        tenant_no="t1",
        contract_id="contract-1",
        contract_no="P0-2026-10-0016",
        user_id="u1",
    )
    db.flush()
    first_updated = db.get(m.PlPlanContract, "link-1").update_time
    changed = sync_plan_contract_no_snapshots(
        db,
        tenant_no="t1",
        contract_id="contract-1",
        contract_no="P0-2026-10-0016",
        user_id="u2",
    )
    db.flush()
    db.expire_all()
    assert changed == 0
    assert db.get(m.PlPlanContract, "link-1").update_by == "u1"
    assert db.get(m.PlPlanContract, "link-1").update_time == first_updated


def test_update_contract_calls_sync_when_contract_no_changes():
    import app.crud.crm.contract as contract_crud

    old = MagicMock()
    old.status = "executing"
    old.contract_type = "sale"
    old.contract_no = "SX/X-JT20261009002"
    old.contract_name = "临时"
    old.party_a = "A"
    old.party_b = "B"
    old.supplier_id = "s"
    old.customer_id = "c"
    old.customer_name = "客"
    old.sign_date = date(2026, 10, 9)
    old.id = "contract-1"

    with patch.object(contract_crud, "_get_contract_row_in_scope", return_value=old), patch.object(
        contract_crud, "invalidate_all_approvals"
    ), patch.object(contract_crud, "_validate_sale_purchase_contract_amount"), patch.object(
        contract_crud, "_apply_contract_entity_fields"
    ), patch.object(contract_crud, "assert_products_allowed_for_supplier"), patch.object(
        contract_crud, "list_contract_products", return_value=[]
    ), patch.object(
        contract_crud, "align_contract_no_to_sign_date", return_value="P0-2026-10-0016"
    ), patch.object(
        contract_crud, "contract_no_exists", return_value=False
    ), patch.object(
        contract_crud, "_build_contract_name_from_parties", return_value="正式"
    ), patch.object(
        contract_crud, "get_contract_detail", return_value={"id": "contract-1"}
    ), patch.object(contract_crud, "sync_contract_products"), patch(
        "app.crud.crm.contract.sync_plan_contract_no_snapshots"
    ) as sync_snap, patch(
        "app.crud.crm.contract.sync_approval_notifications"
    ):
        db = MagicMock()
        contract_crud.update_contract(
            db,
            "contract-1",
            tenant_no="t1",
            data={"contractNo": "P0-2026-10-0016"},
            user_id="u1",
        )
        sync_snap.assert_called_once_with(
            db,
            tenant_no="t1",
            contract_id="contract-1",
            contract_no="P0-2026-10-0016",
            user_id="u1",
        )
        assert old.contract_no == "P0-2026-10-0016"