115 lines
4.0 KiB
Python
115 lines
4.0 KiB
Python
"""
|
||
岗位服务单元测试（P4 服务层改造标杆）
|
||
=============================================================
|
||
|
||
验证 P4 改造后的服务层契约：
|
||
- 服务函数「入参数据、出参数据」
|
||
- 业务失败通过 raise BizError 表达（不再 return R.failed）
|
||
- 纯函数与可 mock 路径无需数据库即可测试
|
||
"""
|
||
|
||
import pytest
|
||
|
||
from src.common.utils.model_utils import build_page_data
|
||
from src.core.exceptions import BizError
|
||
from src.modules.system.position import service as position_service
|
||
|
||
|
||
class TestBuildPageData:
|
||
"""分页数据构建（纯函数）"""
|
||
|
||
def test_shape(self):
|
||
data = build_page_data([], 0, 1, 10)
|
||
assert data == {'records': [], 'total': 0, 'size': 10, 'current': 1, 'pages': 0}
|
||
|
||
|
||
class TestAddPosition:
|
||
"""添加岗位"""
|
||
|
||
def test_empty_data_raises_bizerror(self, flask_app):
|
||
"""空数据触发表单验证失败 → BizError（FlaskForm 需应用上下文）"""
|
||
with flask_app.app_context():
|
||
with pytest.raises(BizError):
|
||
position_service.add_position({})
|
||
|
||
|
||
class TestUpdatePosition:
|
||
"""更新岗位"""
|
||
|
||
def test_empty_data_raises_bizerror(self, flask_app):
|
||
with flask_app.app_context():
|
||
with pytest.raises(BizError):
|
||
position_service.update_position({})
|
||
|
||
|
||
class TestUpdateStatus:
|
||
"""状态设置"""
|
||
|
||
def test_empty_data_raises_bizerror(self, flask_app):
|
||
with flask_app.app_context():
|
||
with pytest.raises(BizError):
|
||
position_service.update_position_status({})
|
||
|
||
|
||
class TestDeletePositions:
|
||
"""批量删除"""
|
||
|
||
def test_empty_raises(self):
|
||
"""空 ID 触发 BizError"""
|
||
with pytest.raises(BizError, match="记录ID不存在"):
|
||
position_service.delete_positions('')
|
||
|
||
def test_invalid_ids_raises(self):
|
||
"""非数字 ID 解析为空 → BizError"""
|
||
with pytest.raises(BizError, match="无效的记录ID"):
|
||
position_service.delete_positions('abc')
|
||
|
||
def test_parse_and_delete(self, app, monkeypatch):
|
||
"""ID 解析后调用仓库批量删除，返回删除计数（before_delete 引用校验需 DB 上下文）"""
|
||
from src.modules.system.position.repository import PositionRepository
|
||
|
||
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 = position_service.delete_positions('1,2,3')
|
||
|
||
assert count == 3
|
||
assert captured['ids'] == [1, 2, 3]
|
||
|
||
|
||
class TestPositionList:
|
||
"""岗位列表（无分页）：参数条件筛选 + status=1 硬性过滤"""
|
||
|
||
@pytest.fixture
|
||
def seed_positions(self, app):
|
||
from src.core.database import db
|
||
from src.modules.system.position.repository import PositionRepository
|
||
|
||
with app.app_context():
|
||
PositionRepository.create(name="产品经理", status=1, sort=1)
|
||
PositionRepository.create(name="项目经理", status=1, sort=2)
|
||
PositionRepository.create(name="测试主管", status=0, sort=3)
|
||
db.session.commit()
|
||
|
||
def test_list_no_filter_returns_enabled_only(self, app, seed_positions):
|
||
with app.app_context():
|
||
data = position_service.get_position_list(None)
|
||
assert [d["name"] for d in data] == ["产品经理", "项目经理"]
|
||
|
||
def test_list_filters_by_name(self, app, seed_positions):
|
||
with app.app_context():
|
||
data = position_service.get_position_list({"name": "经理"})
|
||
assert [d["name"] for d in data] == ["产品经理", "项目经理"]
|
||
|
||
def test_list_status_param_is_ignored(self, app, seed_positions):
|
||
# status 为硬性条件，即使传入 status=0 也只返回启用（status=1）记录
|
||
with app.app_context():
|
||
data = position_service.get_position_list({"name": "经理", "status": 0})
|
||
assert [d["name"] for d in data] == ["产品经理", "项目经理"]
|