shanxia2.0/backend/tests/test_db_inventory_plan_lock.py

"""编辑出入库：有计划号必须存在，且产品只能是该计划的产品。"""
from types import SimpleNamespace

import pytest
from fastapi import HTTPException

from app.api.db.inventory import _enforce_inventory_plan_on_edit


class _Query:
    def __init__(self, rows):
        self.rows = rows

    def filter(self, *args, **kwargs):
        return self

    def all(self):
        return self.rows


def _db(rows):
    db = SimpleNamespace()
    db.query = lambda model: _Query(rows)
    return db


def _row(**kwargs):
    data = {
        "tenant_no": "t1",
        "plan_no": None,
        "product_id": "p-old",
        "purchase_plan_id": None,
        "partner_company": "",
        "business_type": "",
    }
    data.update(kwargs)
    return SimpleNamespace(**data)


def _plan(**kwargs):
    data = {
        "id": "plan-1",
        "plan_no": "JH-1",
        "product_id": "p-plan",
        "customer_name": "山峡客户",
        "business_type": "入库",
    }
    data.update(kwargs)
    return SimpleNamespace(**data)


def test_empty_plan_no_skips_product_restriction():
    payload = {"plan_no": "  ", "product_id": "any-product", "purchase_plan_id": "old"}
    _enforce_inventory_plan_on_edit(_db([]), _row(), payload)
    assert payload["plan_no"] is None
    assert payload["purchase_plan_id"] is None
    assert payload["product_id"] == "any-product"


def test_missing_plan_no_is_rejected():
    payload = {"plan_no": "JH-404", "product_id": "p1"}
    with pytest.raises(HTTPException) as exc:
        _enforce_inventory_plan_on_edit(_db([]), _row(), payload)
    assert exc.value.status_code == 400
    assert exc.value.detail == "计划号不存在"


def test_product_must_match_plan():
    payload = {"plan_no": "JH-1", "product_id": "p-other"}
    with pytest.raises(HTTPException) as exc:
        _enforce_inventory_plan_on_edit(_db([_plan()]), _row(), payload)
    assert exc.value.detail == "有计划号时，产品只能选择该计划的产品"


def test_partner_must_match_plan():
    payload = {
        "plan_no": "JH-1",
        "product_id": "p-plan",
        "partner_company": "其他单位",
        "business_type": "入库",
    }
    with pytest.raises(HTTPException) as exc:
        _enforce_inventory_plan_on_edit(_db([_plan()]), _row(), payload)
    assert exc.value.detail == "有计划号时，往来单位必须与该计划一致"


def test_business_type_must_match_plan():
    payload = {
        "plan_no": "JH-1",
        "product_id": "p-plan",
        "partner_company": "山峡客户",
        "business_type": "出库",
    }
    with pytest.raises(HTTPException) as exc:
        _enforce_inventory_plan_on_edit(_db([_plan()]), _row(), payload)
    assert exc.value.detail == "有计划号时，业务类型必须与该计划一致"


def test_matching_plan_binds_product_partner_and_business_type():
    payload = {
        "plan_no": "JH-1",
        "product_id": "p-plan",
        "partner_company": "山峡客户",
        "business_type": "入库",
    }
    _enforce_inventory_plan_on_edit(_db([_plan()]), _row(), payload)
    assert payload["product_id"] == "p-plan"
    assert payload["partner_company"] == "山峡客户"
    assert payload["business_type"] == "入库"
    assert payload["purchase_plan_id"] == "plan-1"
    assert payload["plan_no"] == "JH-1"


def test_payload_without_plan_or_product_is_ignored():
    payload = {"quantity": 1}
    _enforce_inventory_plan_on_edit(_db([]), _row(plan_no="JH-1"), payload)
    assert payload == {"quantity": 1}