90 lines
3.6 KiB
Python
90 lines
3.6 KiB
Python
"""BaseService 分页/固定过滤钩子单元测试。
|
|
|
|
回归：_apply_page_filters 钩子须被 _build_page_query 调用，
|
|
否则 dict/config 项按所属字典/配置的固定过滤会失效。
|
|
"""
|
|
import json
|
|
|
|
from modules.dictionary.dict.models import Dict
|
|
from modules.dictionary.dict_item.models import DictItem
|
|
from modules.dictionary.dict_item.repository import dictitem_repo
|
|
from modules.dictionary.dict_item.service import dict_item_service
|
|
from modules.configuration.config_item.repository import configitem_repo
|
|
from modules.configuration.config_item.service import config_item_service
|
|
from modules.system.level.repository import level_repo
|
|
from modules.system.level.service import level_service
|
|
from utils import dict_util
|
|
|
|
|
|
def _resp_body(resp):
|
|
return json.loads(resp.body)
|
|
|
|
|
|
def test_dict_item_page_filters_by_dict_id(fake_request, sqlite_db):
|
|
dictitem_repo.create(name="性别男", value="1", dict_id=1, sort=1)
|
|
dictitem_repo.create(name="性别女", value="2", dict_id=1, sort=2)
|
|
dictitem_repo.create(name="状态启用", value="1", dict_id=2, sort=1)
|
|
|
|
fake_request.query_params = {"dict_id": 1}
|
|
resp = dict_item_service.get_page(fake_request)
|
|
body = _resp_body(resp)["data"]
|
|
assert len(body["records"]) == 2
|
|
assert all(row["dict_id"] == 1 for row in body["records"])
|
|
|
|
|
|
def test_config_item_page_filters_by_config_id(fake_request, sqlite_db):
|
|
configitem_repo.create(title="站点名", code="site_name", value="v", type="text", config_id=3, sort=1)
|
|
configitem_repo.create(title="LOGO", code="logo", value="x", type="image", config_id=4, sort=1)
|
|
|
|
# type_name 已由数据字典驱动：种入 config_item_type 并清空进程缓存，避免取到空字典
|
|
d = Dict(name="配置项类型", code="config_item_type", sort=1)
|
|
sqlite_db.add(d)
|
|
sqlite_db.flush()
|
|
sqlite_db.add(DictItem(name="单行文本", value="text", dict_id=d.id, sort=1))
|
|
sqlite_db.flush()
|
|
dict_util._dict_cache.clear()
|
|
|
|
fake_request.query_params = {"config_id": 3}
|
|
resp = config_item_service.get_page(fake_request)
|
|
body = _resp_body(resp)["data"]
|
|
assert len(body["records"]) == 1
|
|
assert body["records"][0]["config_id"] == 3
|
|
# 列表行序列化带 type_name，详情不含
|
|
assert body["records"][0]["type_name"] == "单行文本"
|
|
|
|
|
|
def test_page_like_eq_order(fake_request, sqlite_db):
|
|
level_repo.create(name="初级", status=1, sort=3)
|
|
level_repo.create(name="高级", status=1, sort=1)
|
|
level_repo.create(name="资深", status=2, sort=2)
|
|
|
|
# 等值 status=1 + 排序 sort asc
|
|
fake_request.query_params = {"status": 1}
|
|
resp = level_service.get_page(fake_request)
|
|
body = _resp_body(resp)["data"]
|
|
assert [row["name"] for row in body["records"]] == ["高级", "初级"]
|
|
|
|
# 模糊 name 包含"级"（初级、高级匹配，资深不匹配）
|
|
fake_request.query_params = {"name": "级"}
|
|
resp = level_service.get_page(fake_request)
|
|
assert _resp_body(resp)["data"]["total"] == 2
|
|
|
|
|
|
def test_delete_success(fake_request, sqlite_db):
|
|
obj = level_repo.create(name="待删除", status=1, sort=9)
|
|
resp = level_service.delete(str(obj.id))
|
|
body = _resp_body(resp)
|
|
assert body["code"] == 0
|
|
assert body["msg"] == "本次共删除1条数据"
|
|
assert level_repo.get_by_id(obj.id) is None
|
|
|
|
|
|
def test_delete_missing_id_failed(fake_request, sqlite_db):
|
|
resp = level_service.delete("999")
|
|
assert _resp_body(resp)["msg"] == "记录不存在"
|
|
|
|
|
|
def test_delete_empty_ids_failed(fake_request, sqlite_db):
|
|
resp = level_service.delete("")
|
|
assert _resp_body(resp)["msg"] == "记录ID不存在"
|