"""build_plan_product / upsert 应保持 id 稳定,并拒绝外键无效的产品行 id。""" from unittest.mock import MagicMock import pytest from fastapi import HTTPException from app.schemas.pl.plan import PlanProductCreate from app.services import pl_plan_service as svc from app.services.pl_plan_service import ( build_plan_product, resolve_plan_product_id_for_upsert, ) def _minimal_prod(**kwargs) -> PlanProductCreate: data = { "product_id": "crm-p1", "product_no": "P001", "product_name": "甲醇", } data.update(kwargs) return PlanProductCreate(**data) def test_build_plan_product_keeps_row_id(): result = build_plan_product( tenant_no="t1", plan_id="plan-1", plan_type="outbound", prod=_minimal_prod(), order_num=0, user_id="u1", dept_id="d1", product_row_id="pp-1", ) assert result.id == "pp-1" def test_build_plan_product_generates_when_missing(monkeypatch): monkeypatch.setattr(svc, "generate_uuid", lambda: "gen-1") result = build_plan_product( tenant_no="t1", plan_id="plan-1", plan_type="outbound", prod=_minimal_prod(), order_num=0, user_id="u1", dept_id="d1", ) assert result.id == "gen-1" def test_upsert_plan_product_rejects_foreign_id(): with pytest.raises(HTTPException) as ei: resolve_plan_product_id_for_upsert( plan_id="plan-1", payload_id="pp-other-plan", existing_by_id={}, ) assert ei.value.status_code == 422 def test_upsert_plan_product_keeps_payload_id(): existing = MagicMock() existing.id = "pp-keep" existing.plan_id = "plan-1" row_id, row = resolve_plan_product_id_for_upsert( plan_id="plan-1", payload_id="pp-keep", existing_by_id={"pp-keep": existing}, ) assert row_id == "pp-keep" assert row is existing def test_upsert_plan_product_generates_when_missing(monkeypatch): monkeypatch.setattr(svc, "generate_uuid", lambda: "new-pp") row_id, row = resolve_plan_product_id_for_upsert( plan_id="plan-1", payload_id=None, existing_by_id={}, ) assert row_id == "new-pp" assert row is None