287 lines
9.7 KiB
Python
287 lines
9.7 KiB
Python
"""
|
||
服务基类单元测试（P1 服务层骨架）
|
||
=============================================================
|
||
|
||
验证 BaseService 通用 CRUD 骨架：
|
||
- 模板方法流程：add / update / remove / update_status / get_page / get_detail / get_list
|
||
- 表单校验失败 raise BizError
|
||
- 存在性校验（update / update_status 对不存在记录 raise BizError）
|
||
- 覆写钩子（before_add / after_add / before_delete / after_update）按序触发
|
||
- 模块级代理函数形态（保持 router 零改动）
|
||
|
||
DB 相关用例使用 conftest.app（sqlite 内存库），
|
||
表单校验用例使用 conftest.flask_app（最小应用上下文）。
|
||
"""
|
||
|
||
import pytest
|
||
|
||
from src.common.base_service import BaseService
|
||
from src.common.utils.model_utils import build_page_data
|
||
from src.core.exceptions import BizError
|
||
from src.modules.cms.category.repository import CategoryRepository
|
||
from src.modules.cms.category.service import CategoryService
|
||
from src.modules.dictionary.dict_item.repository import DictItemRepository
|
||
from src.modules.dictionary.dict.repository import DictRepository
|
||
from src.modules.dictionary.dict_item.service import DictItemService
|
||
from src.modules.system.position.repository import PositionRepository
|
||
from src.modules.system.position.schema import PositionForm, PositionStatusForm
|
||
|
||
|
||
def _make_position(app, name, sort=0, status=1):
|
||
"""在测试库中构造一条岗位数据并提交"""
|
||
from src.core.database import db
|
||
with app.app_context():
|
||
pid = PositionRepository.create(name=name, sort=sort, status=status).id
|
||
db.session.commit()
|
||
return pid
|
||
|
||
|
||
class DummyService(BaseService):
|
||
"""覆写全部钩子用于断言触发顺序的测试子类"""
|
||
repository = PositionRepository
|
||
form = PositionForm
|
||
status_form = PositionStatusForm
|
||
model_name = "岗位"
|
||
events = []
|
||
|
||
@classmethod
|
||
def before_add(cls, form):
|
||
cls.events.append("before_add")
|
||
|
||
@classmethod
|
||
def after_add(cls, obj):
|
||
cls.events.append("after_add")
|
||
|
||
@classmethod
|
||
def before_update(cls, form, obj):
|
||
cls.events.append("before_update")
|
||
|
||
@classmethod
|
||
def after_update(cls, obj):
|
||
cls.events.append("after_update")
|
||
|
||
@classmethod
|
||
def before_delete(cls, ids):
|
||
cls.events.append("before_delete")
|
||
|
||
@classmethod
|
||
def after_delete(cls, count):
|
||
cls.events.append("after_delete")
|
||
|
||
|
||
@pytest.fixture(autouse=True)
|
||
def _reset_events():
|
||
DummyService.events = []
|
||
yield
|
||
|
||
|
||
class TestGetPage:
|
||
"""分页查询"""
|
||
|
||
def test_shape_and_records(self, app):
|
||
_make_position(app, "岗位甲")
|
||
|
||
with app.app_context():
|
||
data = DummyService.get_page({"pageNo": 1, "pageSize": 10, "name": "岗位"})
|
||
|
||
assert data["total"] == 1
|
||
assert data["size"] == 10
|
||
assert data["current"] == 1
|
||
assert data["records"][0]["name"] == "岗位甲"
|
||
|
||
def test_empty_page(self, app):
|
||
with app.app_context():
|
||
data = DummyService.get_page({"pageNo": 1, "pageSize": 10})
|
||
assert data == {"records": [], "total": 0, "size": 10, "current": 1, "pages": 0}
|
||
|
||
|
||
class TestGetDetail:
|
||
"""详情查询"""
|
||
|
||
def test_detail_returns_dict(self, app):
|
||
pid = _make_position(app, "详情岗位")
|
||
with app.app_context():
|
||
detail = DummyService.get_detail(pid)
|
||
assert detail is not None
|
||
assert detail["id"] == pid
|
||
assert detail["name"] == "详情岗位"
|
||
|
||
def test_detail_missing_returns_none(self, app):
|
||
with app.app_context():
|
||
assert DummyService.get_detail(99999) is None
|
||
|
||
|
||
class TestGetList:
|
||
"""无分页列表"""
|
||
|
||
def test_list_returns_dicts(self, app):
|
||
_make_position(app, "列表一")
|
||
_make_position(app, "列表二")
|
||
with app.app_context():
|
||
items = DummyService.get_list()
|
||
assert [it["name"] for it in items] == ["列表一", "列表二"]
|
||
|
||
|
||
class TestAdd:
|
||
"""新增"""
|
||
|
||
def test_form_invalid_raises(self, flask_app):
|
||
with flask_app.app_context():
|
||
with pytest.raises(BizError, match="岗位名称不能为空"):
|
||
DummyService.add({})
|
||
|
||
def test_add_creates_and_hooks_run(self, app):
|
||
from src.core.database import db
|
||
with app.app_context():
|
||
DummyService.add({"name": "新增岗位", "status": 1, "sort": 3})
|
||
db.session.commit()
|
||
|
||
position = PositionRepository.first_by({"name": "新增岗位"})
|
||
assert position is not None
|
||
assert position.status == 1
|
||
|
||
assert DummyService.events == ["before_add", "after_add"]
|
||
|
||
|
||
class TestUpdate:
|
||
"""更新"""
|
||
|
||
def test_nonexistent_raises(self, app):
|
||
# 表单唯一性校验与存在性校验均查库，需 sqlite 上下文
|
||
with app.app_context():
|
||
with pytest.raises(BizError, match="岗位不存在"):
|
||
DummyService.update({"id": 99999, "name": "不存在", "status": 1})
|
||
|
||
def test_update_changes_and_hooks_run(self, app):
|
||
from src.core.database import db
|
||
pid = _make_position(app, "原名称")
|
||
|
||
with app.app_context():
|
||
DummyService.update({"id": pid, "name": "新名称", "status": 1, "sort": 1})
|
||
db.session.commit()
|
||
db.session.expire_all()
|
||
position = PositionRepository.get_by_id(pid)
|
||
|
||
assert position.name == "新名称"
|
||
assert DummyService.events == ["before_update", "after_update"]
|
||
|
||
|
||
class TestRemove:
|
||
"""批量删除"""
|
||
|
||
def test_invalid_ids_raises(self):
|
||
with pytest.raises(BizError, match="无效的记录ID"):
|
||
DummyService.remove("abc")
|
||
|
||
def test_remove_calls_repo_and_hooks(self, app, monkeypatch):
|
||
pid = _make_position(app, "待删除")
|
||
captured = {}
|
||
|
||
def fake_batch_delete(cls, ids):
|
||
captured["ids"] = ids
|
||
return len(ids)
|
||
|
||
monkeypatch.setattr(PositionRepository, "batch_delete", classmethod(fake_batch_delete))
|
||
|
||
with app.app_context():
|
||
count = DummyService.remove(str(pid))
|
||
|
||
assert count == 1
|
||
assert captured["ids"] == [pid]
|
||
assert DummyService.events == ["before_delete", "after_delete"]
|
||
|
||
|
||
class TestUpdateStatus:
|
||
"""状态设置"""
|
||
|
||
def test_nonexistent_raises(self, app):
|
||
# 存在性校验查库，需 sqlite 上下文
|
||
with app.app_context():
|
||
with pytest.raises(BizError, match="岗位不存在"):
|
||
DummyService.update_status({"id": 99999, "status": 2})
|
||
|
||
def test_status_updated(self, app):
|
||
pid = _make_position(app, "状态岗位", status=1)
|
||
from src.core.database import db
|
||
with app.app_context():
|
||
DummyService.update_status({"id": pid, "status": 2})
|
||
db.session.commit()
|
||
db.session.expire_all()
|
||
assert PositionRepository.get_by_id(pid).status == 2
|
||
|
||
|
||
class TestProxyShape:
|
||
"""模块级代理函数形态：BaseService 方法可直接绑定为模块函数供 router 调用"""
|
||
|
||
def test_method_binds_to_module_level(self):
|
||
assert callable(DummyService.get_page)
|
||
assert callable(DummyService.add)
|
||
assert callable(DummyService.remove)
|
||
# 绑定后调用签名与现有 service 模块级函数一致
|
||
assert build_page_data([], 0, 1, 10)["pages"] == 0
|
||
|
||
|
||
class TestBeforeAddExtraFields:
|
||
"""before_add 校验父记录存在性"""
|
||
|
||
def test_dict_item_add_validates_dict_exists(self, app):
|
||
from src.core.database import db
|
||
with app.app_context():
|
||
dict_obj = DictRepository.create(name="性别", code="gender", sort=0)
|
||
db.session.commit()
|
||
dict_id = dict_obj.id
|
||
|
||
with app.app_context():
|
||
DictItemService.add({"dictId": dict_id, "name": "男", "value": "1", "sort": 1})
|
||
db.session.commit()
|
||
item = DictItemRepository.first_by({"dict_id": dict_id})
|
||
|
||
assert item is not None
|
||
assert item.name == "男"
|
||
|
||
def test_dict_item_add_missing_dict_raises(self, app):
|
||
with app.app_context():
|
||
with pytest.raises(BizError, match="字典不存在"):
|
||
DictItemService.add({"dictId": 99999, "name": "男", "value": "1"})
|
||
|
||
|
||
class TestMigratedModuleHooks:
|
||
"""真实迁移模块的钩子端到端接线（before_delete 子级校验）"""
|
||
|
||
def _make_category(self, app, name, parent_id=0, sort=0, type_=1):
|
||
from src.core.database import db
|
||
with app.app_context():
|
||
cid = CategoryRepository.create(
|
||
category_name=name, category_code=f"code_{name}",
|
||
parent_id=parent_id, sort=sort, type=type_,
|
||
).id
|
||
db.session.commit()
|
||
return cid
|
||
|
||
def test_category_delete_with_children_rejected(self, app):
|
||
parent_id = self._make_category(app, "总公司")
|
||
self._make_category(app, "技术部", parent_id=parent_id)
|
||
|
||
with app.app_context():
|
||
with pytest.raises(BizError, match="存在子分类"):
|
||
CategoryService.remove(str(parent_id))
|
||
|
||
def test_category_delete_leaf_succeeds(self, app):
|
||
parent_id = self._make_category(app, "总公司")
|
||
child_id = self._make_category(app, "技术部", parent_id=parent_id)
|
||
|
||
with app.app_context():
|
||
count = CategoryService.remove(str(child_id))
|
||
assert count == 1
|
||
|
||
def test_category_add_defaults_type(self, app):
|
||
"""前端不传 type → 模型 default=1 兜底（原实现 NOT NULL 必崩）"""
|
||
from src.core.database import db
|
||
with app.app_context():
|
||
CategoryService.add({"categoryName": "默认类型分类", "categoryCode": "default_type"})
|
||
db.session.commit()
|
||
cat = CategoryRepository.first_by({"category_name": "默认类型分类"})
|
||
|
||
assert cat is not None
|
||
assert cat.type == 1
|