""" 费控制度模块专题测试。 解决策略: 1. 函数内 lazy import 避免模块级依赖 2. patch.object 对已导入类的实例方法打桩 3. patch.object 对导入后的模块全局变量打桩（绕过 @patch 字符串路径限制） """ import json from unittest.mock import AsyncMock, MagicMock, patch import pytest pytestmark = pytest.mark.asyncio # ====================== 测试数据 ====================== MOCK_INSTITUTION_ID = "inst_202605150001" MOCK_ENTERPRISE_ID = "ent_202605150001" MOCK_ISSUE_RULE_ID = "rule_202605150001" @pytest.fixture def mock_auth(): """模拟认证上下文""" auth = MagicMock() auth.user.id = 1 auth.tenant_id = "tenant_001" auth.db = AsyncMock() return auth # ====================== 费控制度 Service 测试 ====================== class TestInstitutionCreate: """测试创建费控制度的串联流程""" async def test_create_full_flow_success(self, mock_auth): """正向场景：完整的创建流程（create + scope + issuerule + 本地保存）""" import app.plugin.module_payment.expense.institution.service as svc inst = svc.InstitutionService scope_svc = svc.InstitutionScopeService issuerule_svc = svc.IssueruleService with ( patch.object(inst, "create_institution_service") as mock_create_inst, patch.object(scope_svc, "scope_modify_service") as mock_scope, patch.object(issuerule_svc, "create_issuerule_service") as mock_issuerule, patch.object(svc, "InstitutionCRUD") as mock_crud_cls, ): mock_create_inst.return_value = MagicMock(institution_id=MOCK_INSTITUTION_ID) mock_scope.return_value = {"result": True} mock_issuerule.return_value = {"issue_rule_id": MOCK_ISSUE_RULE_ID} mock_crud_instance = AsyncMock() mock_crud_cls.return_value = mock_crud_instance from alipay.aop.api.domain.AlipayEbppInvoiceInstitutionCreateModel import ( AlipayEbppInvoiceInstitutionCreateModel, ) model = AlipayEbppInvoiceInstitutionCreateModel() model.enterprise_id = MOCK_ENTERPRISE_ID model.institution_name = "测试制度" model.effective = "1" result = await inst.create_institution_full_flow( auth=mock_auth, institution_model=model, enterprise_id=MOCK_ENTERPRISE_ID, scope_data={"adapter_type": "EMPLOYEE_ALL"}, issuerule_data={ "quota_type": "CAP", "issue_type": "ISSUE_MONTH", "issue_amount_value": "1000", "issue_rule_name": "测试制度-发放规则", }, ) assert result["institution_id"] == MOCK_INSTITUTION_ID assert result["scope_modified"] is True assert result["issue_rule_id"] == MOCK_ISSUE_RULE_ID mock_create_inst.assert_called_once() mock_scope.assert_called_once() mock_issuerule.assert_called_once() mock_crud_instance.create.assert_called_once() async def test_create_without_scope_and_issuerule(self, mock_auth): """边界场景：不传 scope 和 issuerule 时只调 create""" import app.plugin.module_payment.expense.institution.service as svc inst = svc.InstitutionService scope_svc = svc.InstitutionScopeService issuerule_svc = svc.IssueruleService with ( patch.object(inst, "create_institution_service") as mock_create_inst, patch.object(scope_svc, "scope_modify_service") as mock_scope, patch.object(issuerule_svc, "create_issuerule_service") as mock_issuerule, patch.object(svc, "InstitutionCRUD") as mock_crud_cls, ): mock_create_inst.return_value = MagicMock(institution_id=MOCK_INSTITUTION_ID) mock_crud_instance = AsyncMock() mock_crud_cls.return_value = mock_crud_instance from alipay.aop.api.domain.AlipayEbppInvoiceInstitutionCreateModel import ( AlipayEbppInvoiceInstitutionCreateModel, ) model = AlipayEbppInvoiceInstitutionCreateModel() model.enterprise_id = MOCK_ENTERPRISE_ID model.institution_name = "无成员无规则制度" result = await inst.create_institution_full_flow( auth=mock_auth, institution_model=model, enterprise_id=MOCK_ENTERPRISE_ID, scope_data=None, issuerule_data=None, ) assert result["institution_id"] == MOCK_INSTITUTION_ID assert result["scope_modified"] is False assert result["issue_rule_id"] is None mock_scope.assert_not_called() mock_issuerule.assert_not_called() async def test_create_rollback_on_failure(self, mock_auth): """异常场景：scope 失败后触发回滚（删除已创建的制度）""" import app.plugin.module_payment.expense.institution.service as svc inst = svc.InstitutionService scope_svc = svc.InstitutionScopeService with ( patch.object(inst, "create_institution_service") as mock_create_inst, patch.object(scope_svc, "scope_modify_service") as mock_scope, ): mock_create_inst.return_value = MagicMock(institution_id=MOCK_INSTITUTION_ID) mock_scope.side_effect = Exception("支付宝接口异常") with patch("asyncio.to_thread") as mock_to_thread: from alipay.aop.api.domain.AlipayEbppInvoiceInstitutionCreateModel import ( AlipayEbppInvoiceInstitutionCreateModel, ) model = AlipayEbppInvoiceInstitutionCreateModel() model.enterprise_id = MOCK_ENTERPRISE_ID model.institution_name = "测试回滚制度" with pytest.raises(Exception, match="支付宝接口异常"): await inst.create_institution_full_flow( auth=mock_auth, institution_model=model, enterprise_id=MOCK_ENTERPRISE_ID, scope_data={"adapter_type": "EMPLOYEE_ALL"}, issuerule_data=None, ) assert mock_to_thread.called, "回滚时应该调用了 asyncio.to_thread" class TestInstitutionDelete: """测试删除费控制度""" async def test_delete_sync_local_db(self, mock_auth): """正向场景：删除成功后同步删除本地记录""" import app.plugin.module_payment.expense.institution.service as svc from alipay.aop.api.domain.AlipayEbppInvoiceInstitutionDeleteModel import ( AlipayEbppInvoiceInstitutionDeleteModel, ) inst = svc.InstitutionService with ( patch.object(inst, "_execute_alipay") as mock_execute, patch.object(svc, "InstitutionCRUD") as mock_crud_cls, ): mock_execute.return_value = json.dumps({ "alipay_ebpp_invoice_institution_delete_response": { "code": "10000", "msg": "Success", } }) mock_obj = MagicMock() mock_obj.id = 1 mock_crud_instance = AsyncMock() mock_crud_instance.get = AsyncMock(return_value=mock_obj) mock_crud_cls.return_value = mock_crud_instance delete_model = AlipayEbppInvoiceInstitutionDeleteModel() delete_model.institution_id = MOCK_INSTITUTION_ID delete_model.enterprise_id = MOCK_ENTERPRISE_ID result = await inst.delete_institution_service(auth=mock_auth, data=delete_model) assert result is not None mock_crud_instance.get.assert_called_with(institution_id=MOCK_INSTITUTION_ID) mock_crud_instance.delete.assert_called_with(ids=[mock_obj.id]) class TestIssueruleConstraint: """测试自动发放规则的参数约束""" async def test_cap_must_be_accumulative(self, mock_auth): """余额类型(CAP)必须为可累计(invalid_mode=1)""" from app.core.exceptions import CustomException import app.plugin.module_payment.expense.institution.service as svc with patch.object(svc.IssueruleService, "_execute_alipay") as mock_execute: with pytest.raises(CustomException, match="余额类型.*必须为可累计"): await svc.IssueruleService.create_issuerule_service( auth=mock_auth, institution_id=MOCK_INSTITUTION_ID, enterprise_id=MOCK_ENTERPRISE_ID, quota_type="CAP", issue_type="ISSUE_MONTH", issue_amount_value="500", invalid_mode=0, ) mock_execute.assert_not_called() async def test_count_cannot_transfer(self, mock_auth): """次卡类型(COUNT)不可转赠(share_mode=0)""" from app.core.exceptions import CustomException import app.plugin.module_payment.expense.institution.service as svc with patch.object(svc.IssueruleService, "_execute_alipay") as mock_execute: with pytest.raises(CustomException, match="次卡类型.*不可转赠"): await svc.IssueruleService.create_issuerule_service( auth=mock_auth, institution_id=MOCK_INSTITUTION_ID, enterprise_id=MOCK_ENTERPRISE_ID, quota_type="COUNT", issue_type="ISSUE_MONTH", issue_amount_value="10", share_mode=1, ) mock_execute.assert_not_called() # ====================== Scope Sync 测试 ====================== class TestScopeSync: """测试员工/部门停用联动""" async def test_remove_employee(self, mock_auth): """解约员工时：从所有引用该员工的制度中移除""" import app.plugin.module_payment.expense.institution.scope_sync as sync import app.plugin.module_payment.expense.institution.service as svc with ( patch.object(sync, "InstitutionCRUD") as mock_crud_cls, patch.object(svc.InstitutionScopeService, "scope_modify_service") as mock_scope_modify, ): mock_inst_1 = MagicMock() mock_inst_1.institution_id = "inst_001" mock_inst_2 = MagicMock() mock_inst_2.institution_id = "inst_002" mock_crud_instance = AsyncMock() mock_crud_instance.list = AsyncMock(return_value=[mock_inst_1, mock_inst_2]) mock_crud_cls.return_value = mock_crud_instance await sync.remove_employee_from_institution_scopes( auth=mock_auth, enterprise_id=MOCK_ENTERPRISE_ID, employee_id="emp_001", ) assert mock_scope_modify.call_count == 2 mock_scope_modify.assert_any_call( auth=mock_auth, institution_id="inst_001", data={ "enterprise_id": MOCK_ENTERPRISE_ID, "adapter_type": "EMPLOYEE_SELECT", "delete_owner_id_list": ["emp_001"], }, ) async def test_remove_department(self, mock_auth): """停用部门时：从所有引用该部门的制度中移除""" import app.plugin.module_payment.expense.institution.scope_sync as sync import app.plugin.module_payment.expense.institution.service as svc with ( patch.object(sync, "InstitutionCRUD") as mock_crud_cls, patch.object(svc.InstitutionScopeService, "scope_modify_service") as mock_scope_modify, ): mock_inst = MagicMock() mock_inst.institution_id = "inst_001" mock_crud_instance = AsyncMock() mock_crud_instance.list = AsyncMock(return_value=[mock_inst]) mock_crud_cls.return_value = mock_crud_instance await sync.remove_department_from_institution_scopes( auth=mock_auth, enterprise_id=MOCK_ENTERPRISE_ID, department_id="dept_001", ) mock_scope_modify.assert_called_once_with( auth=mock_auth, institution_id="inst_001", data={ "enterprise_id": MOCK_ENTERPRISE_ID, "adapter_type": "EMPLOYEE_DEPARTMENT", "delete_owner_id_list": ["dept_001"], }, )