import asyncio from fastapi_modules.fastapi_leaudit.services.impl.ruleConfigServiceImpl import RuleConfigServiceImpl def test_rule_config_source_fields_are_backfilled_from_binding_and_rule_set_meta(): service = RuleConfigServiceImpl() tenant_binding = {"id": 101, "group_id": 11, "rule_set_id": 201, "tenant_code": "MZ"} provincial_binding = {"id": 102, "group_id": 12, "rule_set_id": 202, "tenant_code": "PROVINCIAL"} public_binding = {"id": 103, "group_id": 13, "rule_set_id": 203, "tenant_code": "PUBLIC"} rule_set_map = { 201: { "rule_type": "contract", "rule_name": "Tenant Rule", "current_version_id": 301, "fallback_version_id": None, "has_usable_version": True, "usable_rule_count": 2, "source_rule_set_id": 901, }, 202: { "rule_type": "contract", "rule_name": "Provincial Rule", "current_version_id": 302, "fallback_version_id": None, "has_usable_version": True, "usable_rule_count": 1, "source_rule_set_id": 902, }, 203: { "rule_type": "contract", "rule_name": "Public Rule", "current_version_id": 303, "fallback_version_id": None, "has_usable_version": True, "usable_rule_count": 1, "source_rule_set_id": 903, }, } row = { "group_id": 11, "root_group_id": 1, "document_type_id": 21, "document_type_name": "合同", "main_type": "主类", "subtype": "子类", "entry_module_name": "模块", } async def run(): service._load_effective_binding = fake_load_effective_binding service._load_yaml_text_by_version_id = fake_load_yaml_text_by_version_id service._load_latest_version_id = fake_load_latest_version_id service._load_current_user = fake_load_current_user tenant_pack = await service._build_pack_vo(row, rule_set_map, CurrentUserId=1) provincial_pack = await service._build_pack_vo({**row, "group_id": 12}, rule_set_map, CurrentUserId=1) public_pack = await service._build_pack_vo({**row, "group_id": 13}, rule_set_map, CurrentUserId=1) tenant_summary = service._build_pack_summary_item( row={**row, "group_id": 11}, binding=tenant_binding, rule_set_map=rule_set_map, latest_version_map={}, current_user={"tenant_code": "MZ", "is_global": False}, ) provincial_summary = service._build_pack_summary_item( row={**row, "group_id": 12}, binding=provincial_binding, rule_set_map=rule_set_map, latest_version_map={}, current_user={"tenant_code": "MZ", "is_global": False}, ) public_summary = service._build_pack_summary_item( row={**row, "group_id": 13}, binding=public_binding, rule_set_map=rule_set_map, latest_version_map={}, current_user={"tenant_code": "MZ", "is_global": False}, ) return tenant_pack, provincial_pack, public_pack, tenant_summary, provincial_summary, public_summary async def fake_load_effective_binding(group_id: int, CurrentUserId=None): return { 11: tenant_binding, 12: provincial_binding, 13: public_binding, }[group_id] async def fake_load_yaml_text_by_version_id(version_id: int): return f"yaml-{version_id}" async def fake_load_latest_version_id(rule_set_id: int): return None async def fake_load_current_user(CurrentUserId=None): return {"tenant_code": "MZ", "is_global": False} tenant_pack, provincial_pack, public_pack, tenant_summary, provincial_summary, public_summary = asyncio.run(run()) assert tenant_pack.effectiveTenantCode == "MZ" assert tenant_pack.effectiveScopeType == "TENANT" assert tenant_pack.isInherited is False assert tenant_pack.sourceRuleSetId == 901 assert provincial_pack.effectiveTenantCode == "PROVINCIAL" assert provincial_pack.effectiveScopeType == "PROVINCIAL" assert provincial_pack.isInherited is True assert provincial_pack.sourceRuleSetId == 902 assert public_pack.effectiveTenantCode == "PUBLIC" assert public_pack.effectiveScopeType == "PUBLIC" assert public_pack.isInherited is True assert public_pack.sourceRuleSetId == 903 assert tenant_summary["effectiveTenantCode"] == "MZ" assert tenant_summary["effectiveScopeType"] == "TENANT" assert tenant_summary["isInherited"] is False assert tenant_summary["sourceRuleSetId"] == 901 assert provincial_summary["effectiveTenantCode"] == "PROVINCIAL" assert provincial_summary["effectiveScopeType"] == "PROVINCIAL" assert provincial_summary["isInherited"] is True assert provincial_summary["sourceRuleSetId"] == 902 assert public_summary["effectiveTenantCode"] == "PUBLIC" assert public_summary["effectiveScopeType"] == "PUBLIC" assert public_summary["isInherited"] is True assert public_summary["sourceRuleSetId"] == 903 def test_rule_config_pack_does_not_fallback_to_cross_tenant_rule_set_latest_version(): service = RuleConfigServiceImpl() row = { "group_id": 11, "root_group_id": 1, "document_type_id": 21, "document_type_name": "合同", "main_type": "主类", "subtype": "子类", "entry_module_name": "模块", } cross_tenant_binding = {"id": 101, "group_id": 11, "rule_set_id": 999, "tenant_code": "JY"} latest_version_calls: list[int] = [] async def run(): service._load_effective_binding = fake_load_effective_binding service._load_yaml_text_by_version_id = fake_load_yaml_text_by_version_id service._load_latest_version_id = fake_load_latest_version_id service._load_current_user = fake_load_current_user return await service._build_pack_vo(row, {}, CurrentUserId=1) async def fake_load_effective_binding(group_id: int, CurrentUserId=None): return cross_tenant_binding async def fake_load_yaml_text_by_version_id(version_id: int): return f"yaml-{version_id}" async def fake_load_latest_version_id(rule_set_id: int): latest_version_calls.append(rule_set_id) return 10010 async def fake_load_current_user(CurrentUserId=None): return {"tenant_code": "MZ", "is_global": False} pack = asyncio.run(run()) assert pack.ruleSetId is None assert pack.resolvedVersionId is None assert pack.yamlText == "" assert latest_version_calls == []