wren
175 lines
6.6 KiB
Python
175 lines
6.6 KiB
Python
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 == []
|