shanxia2.0/backend/tests/test_dict_ensure.py

from types import SimpleNamespace
from unittest.mock import MagicMock

import pytest

from app.models.models import SysDictData
from app.services.dict_ensure import (
    DICT_ENSURE_WHITELIST,
    DictEnsureError,
    ensure_dict_data,
    ensure_remove_dict_data,
)


def test_whitelist_contains_crm_types():
    assert "crm_product_gg" in DICT_ENSURE_WHITELIST
    assert "crm_shfs" in DICT_ENSURE_WHITELIST
    assert "crm_product_unit" in DICT_ENSURE_WHITELIST
    assert "crm_customer_park" in DICT_ENSURE_WHITELIST


def test_ensure_rejects_unknown_dict_type():
    db = MagicMock()
    with pytest.raises(DictEnsureError, match="不支持"):
        ensure_dict_data(
            db, tenant_no="t1", dict_type="sys_user_sex", label="男", user_id="u1"
        )


def test_ensure_rejects_empty_label():
    db = MagicMock()
    with pytest.raises(DictEnsureError, match="不能为空"):
        ensure_dict_data(
            db, tenant_no="t1", dict_type="crm_shfs", label="  ", user_id="u1"
        )


def test_ensure_reuses_existing_label():
    db = MagicMock()
    existing = SimpleNamespace(dict_label="自提", dict_value="自提")
    q = MagicMock()
    q.filter.return_value.first.return_value = existing
    db.query.return_value = q

    result = ensure_dict_data(
        db, tenant_no="t1", dict_type="crm_shfs", label="自提", user_id="u1"
    )
    assert result == {"label": "自提", "value": "自提", "created": False}
    db.add.assert_not_called()


def test_ensure_inserts_new_with_value_equals_label():
    db = MagicMock()
    q = MagicMock()
    # existing by label: None; clash by value: None
    q.filter.return_value.first.side_effect = [None, None]
    db.query.return_value = q

    result = ensure_dict_data(
        db, tenant_no="t1", dict_type="crm_product_gg", label="吨袋装", user_id="u1"
    )
    assert result == {"label": "吨袋装", "value": "吨袋装", "created": True}
    db.add.assert_called_once()
    db.flush.assert_called_once()
    db.commit.assert_not_called()
    added = db.add.call_args[0][0]
    assert isinstance(added, SysDictData)
    assert added.dict_type == "crm_product_gg"
    assert added.dict_label == "吨袋装"
    assert added.dict_value == "吨袋装"
    assert added.tenant_no == "t1"
    assert added.create_by == "u1"


def test_ensure_reuses_existing_by_preferred_value():
    """label miss, but preferred value already exists → reuse, no insert"""
    db = MagicMock()
    existing = SimpleNamespace(dict_label="旧标签", dict_value="吨袋装")
    q = MagicMock()
    # first query by label → None; second by value → existing
    q.filter.return_value.first.side_effect = [None, existing]
    db.query.return_value = q

    result = ensure_dict_data(
        db, tenant_no="t1", dict_type="crm_product_gg", label="吨袋装", user_id="u1"
    )
    assert result == {"label": "旧标签", "value": "吨袋装", "created": False}
    db.add.assert_not_called()


def test_ensure_honors_explicit_preferred_value_on_insert():
    db = MagicMock()
    q = MagicMock()
    q.filter.return_value.first.side_effect = [None, None]
    db.query.return_value = q

    result = ensure_dict_data(
        db,
        tenant_no="t1",
        dict_type="crm_product_gg",
        label="吨袋装",
        user_id="u1",
        preferred_value="custom_gg",
    )
    assert result == {"label": "吨袋装", "value": "custom_gg", "created": True}
    added = db.add.call_args[0][0]
    assert added.dict_value == "custom_gg"


def test_ensure_remove_soft_deletes_own_row():
    db = MagicMock()
    row = SimpleNamespace(
        dict_value="吨袋装",
        create_by="u1",
        del_flag=0,
        update_by=None,
        update_time=None,
    )
    q = MagicMock()
    q.filter.return_value.first.return_value = row
    db.query.return_value = q

    ensure_remove_dict_data(
        db,
        tenant_no="t1",
        dict_type="crm_product_gg",
        value="吨袋装",
        user_id="u1",
    )
    assert row.del_flag == 1
    assert row.update_by == "u1"
    db.add.assert_called_once_with(row)
    db.flush.assert_called_once()


def test_ensure_remove_rejects_other_users_row():
    db = MagicMock()
    row = SimpleNamespace(dict_value="吨袋装", create_by="other", del_flag=0)
    q = MagicMock()
    q.filter.return_value.first.return_value = row
    db.query.return_value = q

    with pytest.raises(DictEnsureError, match="无权|不能删除|不允许"):
        ensure_remove_dict_data(
            db,
            tenant_no="t1",
            dict_type="crm_product_gg",
            value="吨袋装",
            user_id="u1",
        )


def test_ensure_remove_rejects_unknown_type():
    db = MagicMock()
    with pytest.raises(DictEnsureError, match="不支持"):
        ensure_remove_dict_data(
            db,
            tenant_no="t1",
            dict_type="sys_user_sex",
            value="1",
            user_id="u1",
        )