签入版本

tests/unit/test_common/test_base_repository.py

@@ -0,0 +1,392 @@
+"""
+仓储基类单元测试（P0 基建）
+=============================================================
+
+使用 sqlite 内存库（conftest.app 夹具）验证 BaseRepository 通用能力：
+- find_page/find_list 的 like 通配符转义
+- 多字段排序（order_by + order_desc 单次合并，修复覆盖 bug）
+- 软删除与恢复（batch_delete 后 get_by_id 不可见）
+- update_by_id 自动补齐 update_time/update_user（显式值优先）
+- first_by 等值单条查询 / exists / create
+- calc_pages / build_page_data 分页结构
+"""
+
+from datetime import datetime
+
+from src.core.database import db
+from src.common.constants.common import DEFAULT_PASSWORD
+from src.core.password import encrypt_password, generate_salt
+from src.modules.system.position.repository import PositionRepository
+from src.modules.system.user.repository import UserRepository
+from src.modules.system.role.repository import RoleRepository
+from src.modules.dictionary.dict.repository import DictRepository
+from src.modules.dictionary.dict_item.repository import DictItemRepository
+from src.modules.association.user_role.repository import UserRoleRepository
+from src.modules.operation_log.repository import OperationLogRepository
+from src.common.utils.model_utils import calc_pages, build_page_data
+
+
+def _make_position(app, name, sort=0, status=1, create_time=None):
+    """在测试库中构造一条岗位数据并提交"""
+    data = {'name': name, 'sort': sort, 'status': status}
+    if create_time is not None:
+        data['create_time'] = create_time
+    with app.app_context():
+        position = PositionRepository.create(**data)
+        db.session.commit()
+        return position.id
+
+
+def _names(app, items):
+    return [p.name for p in items]
+
+
+class TestFindPageLike:
+    """like 模糊查询通配符转义"""
+
+    def test_special_chars_escaped(self, app):
+        _make_position(app, '100%_x')
+        _make_position(app, '100_x')
+        _make_position(app, '经理100%')
+        _make_position(app, '普通')
+
+        with app.app_context():
+            items = PositionRepository.find_list(like_filters={'name': '100%_x'})
+
+        # 通配符被转义，仅匹配字面量 "100%_x"
+        assert _names(app, items) == ['100%_x']
+
+
+class TestEmptyFilterValues:
+    """空筛选值处理：None 或 '' 不参与筛选（修复前端传 '' 导致查不出数据）"""
+
+    def test_empty_string_equality_ignored(self, app):
+        _make_position(app, '甲')
+        _make_position(app, '乙')
+        with app.app_context():
+            items = PositionRepository.find_list(filters={'name': ''})
+        assert len(items) == 2
+
+    def test_none_equality_ignored(self, app):
+        _make_position(app, '甲')
+        _make_position(app, '乙')
+        with app.app_context():
+            items = PositionRepository.find_list(filters={'name': None})
+        assert len(items) == 2
+
+    def test_zero_equality_still_filters(self, app):
+        _make_position(app, '启用', status=1)
+        _make_position(app, '禁用', status=0)
+        with app.app_context():
+            items = PositionRepository.find_list(filters={'status': 0})
+        assert len(items) == 1
+
+    def test_empty_string_like_ignored(self, app):
+        _make_position(app, '甲')
+        _make_position(app, '乙')
+        with app.app_context():
+            items = PositionRepository.find_list(like_filters={'name': ''})
+        assert len(items) == 2
+
+
+class TestMultiFieldOrder:
+    """多字段排序：order_by 与 order_desc 单次合并，不再相互覆盖"""
+
+    def test_sort_then_id_desc(self, app):
+        _make_position(app, 'A', sort=2)
+        _make_position(app, 'B', sort=1)
+        _make_position(app, 'C', sort=1)
+        _make_position(app, 'D', sort=2)
+
+        with app.app_context():
+            items = PositionRepository.find_list(order_by=['sort'], order_desc=['id'])
+
+        # sort 升序（1 组在前），组内 id 降序
+        assert _names(app, items) == ['C', 'B', 'D', 'A']
+
+
+class TestSoftDelete:
+    """软删除与恢复"""
+
+    def test_batch_delete_hides_records(self, app):
+        id1 = _make_position(app, '甲')
+        id2 = _make_position(app, '乙')
+        id3 = _make_position(app, '丙')
+
+        with app.app_context():
+            count = PositionRepository.batch_delete([id1, id2])
+            assert count == 2
+            assert PositionRepository.get_by_id(id1) is None
+            assert PositionRepository.get_by_id(id2) is None
+            assert PositionRepository.get_by_id(id3) is not None
+            assert len(PositionRepository.get_all()) == 1
+
+    def test_batch_restore_recovers(self, app):
+        id1 = _make_position(app, '甲')
+        id2 = _make_position(app, '乙')
+
+        with app.app_context():
+            PositionRepository.batch_delete([id1])
+            restored = PositionRepository.batch_restore([id1])
+            assert restored == 1
+            assert PositionRepository.get_by_id(id1) is not None
+            assert PositionRepository.get_by_id(id2) is not None
+
+
+class TestUpdate:
+    """更新能力"""
+
+    def test_update_by_id_stamps_and_keeps_explicit(self, app):
+        pid = _make_position(app, '原名称')
+
+        with app.app_context():
+            affected = PositionRepository.update_by_id(pid, {'name': '新名称'})
+            assert affected == 1
+            db.session.expire_all()
+            position = PositionRepository.get_by_id(pid)
+            assert position.name == '新名称'
+            assert position.update_time is not None
+
+    def test_update_by_id_explicit_values_preferred(self, app):
+        pid = _make_position(app, '原名称')
+
+        with app.app_context():
+            PositionRepository.update_by_id(pid, {'name': '显式更新', 'update_user': 'admin'})
+            db.session.expire_all()
+            position = PositionRepository.get_by_id(pid)
+            assert position.name == '显式更新'
+            assert position.update_user == 'admin'
+
+    def test_update_status(self, app):
+        pid = _make_position(app, '岗位')
+
+        with app.app_context():
+            affected = PositionRepository.update_status(pid, 0)
+            assert affected == 1
+            db.session.expire_all()
+            assert PositionRepository.get_by_id(pid).status == 0
+
+
+class TestFirstByAndExists:
+    """等值单条查询与存在性检查"""
+
+    def test_first_by(self, app):
+        _make_position(app, '唯一岗位')
+        with app.app_context():
+            found = PositionRepository.first_by({'name': '唯一岗位'})
+            assert found is not None
+            assert PositionRepository.first_by({'name': '不存在'}) is None
+
+    def test_exists(self, app):
+        pid = _make_position(app, '存在性检查')
+        with app.app_context():
+            assert PositionRepository.exists(pid) is True
+            assert PositionRepository.exists(99999) is False
+
+
+class TestExistsBy:
+    """唯一性校验 exists_by（等值条件 + 更新时排除自身）"""
+
+    def test_hit_and_miss(self, app):
+        _make_position(app, '唯一岗位')
+        with app.app_context():
+            assert PositionRepository.exists_by({'name': '唯一岗位'}) is True
+            assert PositionRepository.exists_by({'name': '不存在的岗位'}) is False
+
+    def test_exclude_id(self, app):
+        pid = _make_position(app, '同名岗位')
+        with app.app_context():
+            # 更新场景：排除自身后不应命中
+            assert PositionRepository.exists_by({'name': '同名岗位'}, exclude_id=pid) is False
+
+    def test_ignores_soft_deleted(self, app):
+        pid = _make_position(app, '软删岗位')
+        with app.app_context():
+            PositionRepository.batch_delete([pid])
+            assert PositionRepository.exists_by({'name': '软删岗位'}) is False
+
+    def test_compound_filters(self, app):
+        with app.app_context():
+            dict_obj = DictRepository.create(name='性别', code='gender', sort=0)
+            db.session.commit()
+            dict_id = dict_obj.id
+            DictItemRepository.create(dict_id=dict_id, name='男', value='1', sort=1)
+            db.session.commit()
+
+            assert DictItemRepository.exists_by({'dict_id': dict_id, 'value': '1'}) is True
+            assert DictItemRepository.exists_by({'dict_id': dict_id, 'value': '2'}) is False
+
+    def test_empty_filters_false(self, app):
+        with app.app_context():
+            assert PositionRepository.exists_by({}) is False
+            assert PositionRepository.exists_by({'not_a_column': 1}) is False
+
+
+class TestCreate:
+    """创建能力"""
+
+    def test_create_returns_object(self, app):
+        with app.app_context():
+            position = PositionRepository.create(name='新建岗位', sort=5)
+            db.session.commit()
+            assert position.id is not None
+            assert PositionRepository.get_by_id(position.id).name == '新建岗位'
+
+    def test_get_by_ids(self, app):
+        id1 = _make_position(app, '一')
+        id2 = _make_position(app, '二')
+        with app.app_context():
+            records = PositionRepository.get_by_ids([id1, id2, 99999])
+            assert len(records) == 2
+
+
+def _make_operation_log(app, title, create_time):
+    """在测试库中构造一条操作日志并提交（覆盖 create_time）"""
+    with app.app_context():
+        log = OperationLogRepository.create(
+            title=title,
+            type=1,
+            source=0,
+            method='test',
+            request_method='GET',
+            url='/api/v1/test',
+            param='',
+            result='',
+            ip='127.0.0.1',
+            location='',
+            os='',
+            browser='',
+            consume_time=0,
+            status=0,
+            error='',
+            create_time=create_time,
+        )
+        db.session.commit()
+        return log.id
+
+
+class TestLogTimeRange:
+    """操作日志时间范围筛选（page 参数映射）"""
+
+    def test_page_time_range_filter(self, app):
+        _make_operation_log(app, '一月日志', datetime(2026, 1, 15, 10, 0, 0))
+        _make_operation_log(app, '二月日志', datetime(2026, 2, 15, 10, 0, 0))
+        _make_operation_log(app, '三月日志', datetime(2026, 3, 15, 10, 0, 0))
+
+        with app.app_context():
+            items, total, pages = OperationLogRepository.page({
+                'pageNo': 1,
+                'pageSize': 10,
+                'startTime': '2026-02-01 00:00:00',
+                'endTime': '2026-02-28 23:59:59',
+            })
+
+        assert total == 1
+        assert pages == 1
+        assert items[0].title == '二月日志'
+
+    def test_invalid_date_ignored(self, app):
+        _make_operation_log(app, '日志', datetime(2026, 1, 15, 10, 0, 0))
+
+        with app.app_context():
+            items, total, _ = OperationLogRepository.page({
+                'pageNo': 1,
+                'pageSize': 10,
+                'startTime': 'not-a-date',
+            })
+
+        assert total == 1
+        assert items[0].title == '日志'
+
+
+class TestPageDataUtils:
+    """分页工具函数"""
+
+    def test_calc_pages(self):
+        assert calc_pages(0, 10) == 0
+        assert calc_pages(5, 10) == 1
+        assert calc_pages(10, 10) == 1
+        assert calc_pages(11, 10) == 2
+
+    def test_build_page_data(self):
+        data = build_page_data([], 0, 1, 10)
+        assert data == {'records': [], 'total': 0, 'size': 10, 'current': 1, 'pages': 0}
+
+    def test_build_page_data_with_enrich(self, app):
+        _make_position(app, '富化')
+        with app.app_context():
+            items = PositionRepository.find_list()
+            data = build_page_data(
+                items, len(items), 1, 10, enrich=lambda records: [
+                    {**r, 'extra': 'added'} for r in records
+                ]
+            )
+        assert data['total'] == 1
+        assert data['pages'] == 1
+        assert data['records'][0]['extra'] == 'added'
+
+
+def _make_user(app, username):
+    """在测试库中构造一条用户数据并提交（补齐 NOT NULL 字段）"""
+    salt = generate_salt()
+    with app.app_context():
+        user = UserRepository.create(
+            username=username,
+            realname=username,
+            avatar='',
+            mobile='13800000000',
+            email='test@test.com',
+            province_code='',
+            city_code='',
+            district_code='',
+            street_code='',
+            address='',
+            password=encrypt_password(DEFAULT_PASSWORD, salt),
+            salt=salt,
+        )
+        db.session.commit()
+        return user.id
+
+
+def _make_role(app, name, code):
+    """在测试库中构造一条角色数据并提交"""
+    with app.app_context():
+        role = RoleRepository.create(name=name, code=code)
+        db.session.commit()
+        return role.id
+
+
+class TestReplaceRoles:
+    """用户角色关联 replace_roles 先删后增"""
+
+    def test_replace_roles_delete_then_insert(self, app):
+        user_id = _make_user(app, 'test_user')
+        role1 = _make_role(app, '管理员', 'admin')
+        role2 = _make_role(app, '普通用户', 'common')
+
+        with app.app_context():
+            inserted = UserRoleRepository.replace_roles(user_id, [role1, role2])
+            assert inserted == 2
+            db.session.commit()
+
+        # 二次替换：旧关联全部删除，仅保留 role2
+        with app.app_context():
+            inserted = UserRoleRepository.replace_roles(user_id, [role2])
+            assert inserted == 1
+            db.session.commit()
+
+            role_ids = sorted(r.id for r in UserRoleRepository.get_roles_by_user(user_id))
+            assert role_ids == [role2]
+
+    def test_replace_roles_empty_clears_all(self, app):
+        user_id = _make_user(app, 'test_user2')
+        role1 = _make_role(app, '管理员', 'admin')
+
+        with app.app_context():
+            UserRoleRepository.replace_roles(user_id, [role1])
+            db.session.commit()
+            assert len(UserRoleRepository.get_roles_by_user(user_id)) == 1
+
+            UserRoleRepository.replace_roles(user_id, [])
+            db.session.commit()
+            assert UserRoleRepository.get_roles_by_user(user_id) == []

tests/unit/test_common/test_base_service.py

@@ -0,0 +1,286 @@
+"""
+服务基类单元测试（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

tests/unit/test_common/test_database.py

@@ -0,0 +1,201 @@
+"""
+数据库 URL 构建单元测试（D1 多库支持）
+=============================================================
+
+验证 _build_database_url() 对 mysql / postgresql / mssql / sqlite / oracle
+五类驱动的 URL 构建正确，且 sqlite 无 host/port/账号分支。
+"""
+
+import config.database as db_mod
+from src.common.constants.common import DEFAULT_PASSWORD
+
+
+def _url(monkeypatch, driver, **overrides):
+    defaults = {
+        'DB_DRIVER': driver,
+        'DB_HOST': '127.0.0.1',
+        'DB_PORT': 1,
+        'DB_DATABASE': 'appdb',
+        'DB_USERNAME': 'user',
+        'DB_PASSWORD': 'pass',
+        'DB_SSL': False,
+    }
+    defaults.update(overrides)
+    for key, value in defaults.items():
+        monkeypatch.setattr(db_mod, key, value)
+    return db_mod._build_database_url()
+
+
+class TestBuildDatabaseUrl:
+    def test_mysql_default(self, monkeypatch):
+        assert _url(monkeypatch, 'mysql') == 'mysql+pymysql://user:pass@127.0.0.1:1/appdb?charset=utf8mb4'
+
+    def test_postgresql(self, monkeypatch):
+        assert _url(monkeypatch, 'postgresql') == 'postgresql+psycopg2://user:pass@127.0.0.1:1/appdb'
+
+    def test_pgsql_alias(self, monkeypatch):
+        assert _url(monkeypatch, 'pgsql') == 'postgresql+psycopg2://user:pass@127.0.0.1:1/appdb'
+
+    def test_mssql(self, monkeypatch):
+        assert _url(monkeypatch, 'mssql') == 'mssql+pymssql://user:pass@127.0.0.1:1/appdb'
+
+    def test_sqlite_file_path(self, monkeypatch):
+        # Windows 反斜杠路径应规范化为正斜杠，且无 host/账号
+        assert _url(monkeypatch, 'sqlite', DB_DATABASE=r'D:\data\app.db') == 'sqlite:///D:/data/app.db'
+
+    def test_sqlite_memory(self, monkeypatch):
+        assert _url(monkeypatch, 'sqlite', DB_DATABASE=':memory:') == 'sqlite://'
+
+    def test_oracle_service_name(self, monkeypatch):
+        # Oracle 无"数据库名"，DB_DATABASE 即 SERVICE_NAME，走 ?service_name= 查询参数
+        assert _url(monkeypatch, 'oracle') == 'oracle+oracledb://user:pass@127.0.0.1:1/?service_name=appdb'
+
+    def test_unknown_driver_falls_back_mysql(self, monkeypatch):
+        assert _url(monkeypatch, 'db2') == 'mysql+pymysql://user:pass@127.0.0.1:1/appdb?charset=utf8mb4'
+
+    def test_driver_default_ports(self):
+        # DB_PORT 未显式配置时按驱动回退（_DEFAULT_PORT 表）
+        assert db_mod._DEFAULT_PORT['oracle'] == 1521
+        assert db_mod._DEFAULT_PORT['mysql'] == 3306
+        assert db_mod._DEFAULT_PORT['postgresql'] == 5432
+        assert db_mod._DEFAULT_PORT['pgsql'] == 5432
+        assert db_mod._DEFAULT_PORT['mssql'] == 1433
+        assert db_mod._DEFAULT_PORT['sqlite'] == 0
+
+
+class TestSeedData:
+    """种子数据（D3）：幂等写入 admin/角色/菜单"""
+
+    def test_seed_creates_admin_role_menu(self, app):
+        from scripts.seed_data import seed_all
+        from src.modules.system.user.repository import UserRepository
+        from src.modules.system.role.repository import RoleRepository
+        from src.modules.system.menu.repository import MenuRepository
+
+        with app.app_context():
+            seed_all()
+            admin = UserRepository.first_by({'username': 'admin'})
+            assert admin is not None
+            assert admin.password != DEFAULT_PASSWORD  # 已加密
+            assert admin.salt
+            role = RoleRepository.first_by({'code': 'admin'})
+            assert role is not None
+            assert MenuRepository.first_by({'name': '系统管理', 'parent_id': 0}) is not None
+
+    def test_seed_is_idempotent(self, app):
+        from scripts.seed_data import seed_all
+        from src.modules.system.user.repository import UserRepository
+
+        with app.app_context():
+            seed_all()
+            seed_all()
+            admins = UserRepository.find_list(filters={'username': 'admin'})
+            assert len(admins) == 1
+
+
+class TestNormalizeDriver:
+    """驱动别名归一化（_normalize_driver，与 migrate_db.py 共用）"""
+
+    def test_canonical(self):
+        from config.database import _normalize_driver
+        assert _normalize_driver('mysql') == 'mysql'
+        assert _normalize_driver('postgresql') == 'postgresql'
+        assert _normalize_driver('mssql') == 'mssql'
+        assert _normalize_driver('sqlite') == 'sqlite'
+        assert _normalize_driver('oracle') == 'oracle'
+
+    def test_aliases(self):
+        from config.database import _normalize_driver
+        assert _normalize_driver('postgres') == 'postgresql'
+        assert _normalize_driver('pgsql') == 'postgresql'
+        assert _normalize_driver('sqlserver') == 'mssql'
+        assert _normalize_driver('sqlite3') == 'sqlite'
+
+    def test_upper_case(self):
+        from config.database import _normalize_driver
+        assert _normalize_driver('PostgreSQL') == 'postgresql'
+
+
+class TestSqliteUrl:
+    """SQLite 连接 URL 构建（_sqlite_url，与 migrate_db.py 共用）"""
+
+    def test_memory(self):
+        from config.database import _sqlite_url
+        assert _sqlite_url(':memory:') == 'sqlite://'
+
+    def test_windows_path(self):
+        from config.database import _sqlite_url
+        assert _sqlite_url(r'D:\data\app.db') == 'sqlite:///D:/data/app.db'
+
+
+class TestMigrateDbUrl:
+    """跨库迁移脚本 migrate_db.py 的 URL 构建（显式参数版，M1）"""
+
+    def test_make_url_mysql(self):
+        from scripts.migrate_db import make_url
+        assert make_url('mysql', '10.0.0.1', 3306, 'src_db', 'root', 'pwd') \
+            == 'mysql+pymysql://root:pwd@10.0.0.1:3306/src_db?charset=utf8mb4'
+
+    def test_make_url_postgresql(self):
+        from scripts.migrate_db import make_url
+        assert make_url('postgresql', '10.0.0.1', 5432, 'dst_db', 'pg', 'pwd') \
+            == 'postgresql+psycopg2://pg:pwd@10.0.0.1:5432/dst_db'
+
+    def test_make_url_oracle_service_name(self):
+        from scripts.migrate_db import make_url
+        assert make_url('oracle', '10.0.0.1', 1521, 'XEPDB1', 'system', 'pwd') \
+            == 'oracle+oracledb://system:pwd@10.0.0.1:1521/?service_name=XEPDB1'
+
+    def test_make_url_mssql(self):
+        from scripts.migrate_db import make_url
+        assert make_url('mssql', '10.0.0.1', 1433, 'dst_db', 'sa', 'pwd') \
+            == 'mssql+pymssql://sa:pwd@10.0.0.1:1433/dst_db?charset=utf8'
+
+    def test_make_url_sqlite_path(self):
+        from scripts.migrate_db import make_url
+        assert make_url('sqlite', '', None, r'D:\data\app.db', '', '') == 'sqlite:///D:/data/app.db'
+
+    def test_make_url_driver_alias(self):
+        from scripts.migrate_db import make_url
+        # 别名：postgres→postgresql、sqlserver→mssql
+        assert make_url('postgres', '10.0.0.1', 5432, 'dst_db', 'pg', 'pwd') \
+            == 'postgresql+psycopg2://pg:pwd@10.0.0.1:5432/dst_db'
+        assert make_url('sqlserver', '10.0.0.1', 1433, 'dst_db', 'sa', 'pwd') \
+            == 'mssql+pymssql://sa:pwd@10.0.0.1:1433/dst_db?charset=utf8'
+
+
+class TestFriendlyDbError:
+    """数据库连接异常友好提示（翻译认证失败/库不存在/拒连）"""
+
+    def test_auth_failed_gbk(self):
+        from config import friendly_db_error
+        gbk = '用户 "postgres" Password 认证失败'.encode('gbk')
+        e = UnicodeDecodeError('utf-8', gbk, 0, len(gbk), 'invalid continuation byte')
+        assert '认证失败' in friendly_db_error(e)
+        assert 'DB_PASSWORD' in friendly_db_error(e)
+
+    def test_db_not_exist(self):
+        from config import friendly_db_error
+        assert '数据库不存在' in friendly_db_error(Exception('database "x" does not exist'))
+
+    def test_connection_refused(self):
+        from config import friendly_db_error
+        assert '连接被拒绝' in friendly_db_error(Exception('Connection refused'))
+
+    def test_oracle_auth_failed(self):
+        from config import friendly_db_error
+        assert '认证失败' in friendly_db_error(Exception('ORA-01017: invalid username/password; logon denied'))
+
+    def test_oracle_service_name_error(self):
+        from config import friendly_db_error
+        msg = friendly_db_error(Exception('ORA-12514: TNS:listener does not currently know of service requested'))
+        assert '服务名' in msg
+        assert 'SERVICE_NAME' in msg
+
+    def test_oracle_connection_refused(self):
+        from config import friendly_db_error
+        assert '连接被拒绝' in friendly_db_error(Exception('ORA-12170: TNS:Connect timeout occurred'))
+
+    def test_unknown_error(self):
+        from config import friendly_db_error
+        assert 'boom' in friendly_db_error(Exception('boom'))

tests/unit/test_common/test_excel_utils.py

@@ -0,0 +1,145 @@
+"""
+共享 Excel 工具单元测试（excel_utils）
+=============================================================
+
+验证通用 Excel 导入/导出机制（供职级/用户等模块复用）：
+    - parse_excel_rows：按列定义读取，必填/解析错误逐行收集，空行跳过
+    - parse_int / parse_status / format_status：单元格解析与状态展示格式化
+    - build_export_rows：按列 key 取值并应用 formatter
+    - save_export_file：写工作簿、生成访问 URL（路径通过 monkeypatch 指向临时目录）
+
+不依赖具体业务模块，列定义在用例内自构造。
+"""
+
+import pytest
+from openpyxl.reader.excel import load_workbook
+from openpyxl.workbook import Workbook
+
+from src.common.utils import excel_utils
+
+
+# 用例自构造的列定义（结构与 level/user 模块一致）
+COLUMNS = [
+    {"title": "名称", "index": 1, "key": "name", "required": True},
+    {"title": "排序", "index": 2, "key": "sort", "parser": excel_utils.parse_int},
+    {"title": "状态", "index": 3, "key": "status", "parser": excel_utils.parse_status},
+]
+
+
+class TestParseStatus:
+    """状态单元格解析：中文 / 数字 -> 1/2，非法值报错"""
+
+    def test_normal_and_disabled(self):
+        assert excel_utils.parse_status("正常") == 1
+        assert excel_utils.parse_status("禁用") == 2
+        assert excel_utils.parse_status("1") == 1
+        assert excel_utils.parse_status(2) == 2
+
+    def test_invalid_raises(self):
+        with pytest.raises(ValueError):
+            excel_utils.parse_status("未知")
+
+
+class TestParseInt:
+    """整数单元格解析"""
+
+    def test_valid(self):
+        assert excel_utils.parse_int("10") == 10
+        assert excel_utils.parse_int(3) == 3
+
+    def test_invalid_raises(self):
+        with pytest.raises(ValueError):
+            excel_utils.parse_int("abc")
+
+
+class TestFormatStatus:
+    """状态展示格式化：1/2 -> 正常/禁用，未知值原样返回"""
+
+    def test_mapping(self):
+        assert excel_utils.format_status(1) == "正常"
+        assert excel_utils.format_status(2) == "禁用"
+
+    def test_unknown_passthrough(self):
+        assert excel_utils.format_status(0) == 0
+
+
+class TestBuildExportRows:
+    """导出行构建：按列 key 取值 + formatter 展示格式化"""
+
+    def test_formatter_applied_and_key_lookup(self):
+        records = [
+            {"name": "初级", "status": 1, "sort": 2},
+            {"name": "禁用级", "status": 2, "sort": 1},
+        ]
+        columns = [
+            {"title": "名称", "key": "name"},
+            {"title": "排序", "key": "sort"},
+            {"title": "状态", "key": "status", "formatter": excel_utils.format_status},
+        ]
+        assert excel_utils.build_export_rows(records, columns) == [
+            ["初级", 2, "正常"],
+            ["禁用级", 1, "禁用"],
+        ]
+
+    def test_missing_key_becomes_none(self):
+        columns = [{"title": "缺失键", "key": "nope"}]
+        assert excel_utils.build_export_rows([{"name": "x"}], columns) == [[None]]
+
+
+class TestParseExcelRows:
+    """导入行解析：必填校验 / parser 错误 / 空行跳过 / 可选空单元格"""
+
+    def _make_wb(self, path):
+        wb = Workbook()
+        ws = wb.worksheets[0]
+        ws.append(["名称", "排序", "状态"])  # 表头（start_row=2 跳过）
+        ws.append(["初级", 1, "正常"])        # 合法
+        ws.append(["坏排序", "abc", "正常"])  # 排序解析失败
+        ws.append(["缺状态", 2, ""])          # 可选空单元格 -> None
+        ws.append([None, None, None])         # 整行为空 -> 跳过
+        ws.append(["", 3, "禁用"])             # 必填缺失
+        wb.save(path)
+
+    def test_rows_and_errors(self, tmp_path):
+        path = tmp_path / "in.xlsx"
+        self._make_wb(path)
+
+        rows, errors = excel_utils.parse_excel_rows(str(path), COLUMNS)
+
+        # 合法行 + 可选空单元格行保留；错误行 / 空行被剔除
+        assert rows == [
+            {"name": "初级", "sort": 1, "status": 1},
+            {"name": "缺状态", "sort": 2, "status": None},
+        ]
+        # 行级错误带行号与原因
+        assert any("第3行 排序：" in e for e in errors)
+        assert any("第6行 名称不能为空" in e for e in errors)
+
+
+class TestSaveExportFile:
+    """导出文件落盘与 URL 生成"""
+
+    def test_writes_file_and_returns_url(self, tmp_path, monkeypatch):
+        monkeypatch.setattr(excel_utils, "FLASK_TEMP_PATH", str(tmp_path))
+        monkeypatch.setattr(excel_utils, "get_file_url", lambda url: "/static" + url)
+
+        headers = ["名称", "状态"]
+        rows = [["初级", "正常"], ["高级", "禁用"]]
+        url = excel_utils.save_export_file(headers, rows, filename_prefix="level")
+
+        # URL 由 get_file_url 拼接（返回约定前缀 + 路径）
+        assert url.startswith("/static")
+
+        # 文件真实落盘到临时目录，表头与数据一致
+        import glob
+
+        saved = glob.glob(str(tmp_path) + "/**/*.xlsx", recursive=True)
+        assert len(saved) == 1
+        wb = load_workbook(saved[0])
+        ws = wb.worksheets[0]
+        assert list(ws.iter_rows(values_only=True)) == [
+            ("名称", "状态"),
+            ("初级", "正常"),
+            ("高级", "禁用"),
+        ]
+        wb.close()

tests/unit/test_common/test_rich_text.py

@@ -0,0 +1,47 @@
+"""
+save_file 安全修复测试
+=============================================================
+
+验证临时文件迁移的防护与正式路径保留：
+    - FLASK_FILE_URL 为空时拒绝外链（原 find('')==0 恒真，外链被当本地文件）
+    - URL 前缀不匹配时拒绝
+    - 路径穿越（.. 逃逸临时目录）拒绝
+    - 正式路径（不含 /temp）原样保留：相对路径（如 /images/notice/...）编辑回传
+      不被清空，带前缀的绝对地址只去前缀，外链绝对地址仍拒绝
+"""
+
+from src.common.utils import rich_text
+
+
+def test_external_url_rejected_when_prefix_empty(monkeypatch):
+    monkeypatch.setattr(rich_text, 'FLASK_FILE_URL', '')
+    assert rich_text.save_file('http://evil.com/x.jpg', 'user') == ""
+
+
+def test_external_url_nonmatching_prefix_rejected(monkeypatch):
+    monkeypatch.setattr(rich_text, 'FLASK_FILE_URL', 'http://img.site')
+    assert rich_text.save_file('http://evil.com/x.jpg', 'user') == ""
+
+
+def test_path_traversal_rejected(monkeypatch):
+    monkeypatch.setattr(rich_text, 'FLASK_FILE_URL', 'http://img.site')
+    # realpath 规范化后逃逸临时目录 → 拒绝
+    assert rich_text.save_file('http://img.site/temp/../../../../../../etc/passwd', 'user') == ""
+
+
+def test_non_temp_url_returns_stripped_path(monkeypatch):
+    monkeypatch.setattr(rich_text, 'FLASK_FILE_URL', 'http://img.site')
+    # 已是正式路径（无 /temp）→ 去掉前缀返回，不迁移
+    assert rich_text.save_file('http://img.site/user/avatar.jpg', 'user') == "/user/avatar.jpg"
+
+
+def test_non_temp_relative_path_kept_when_prefix_mismatch(monkeypatch):
+    monkeypatch.setattr(rich_text, 'FLASK_FILE_URL', 'http://img.site')
+    # 编辑回传已保存的相对路径（如通知封面），前缀不匹配也应原样保留，不置空
+    assert rich_text.save_file('/images/notice/20260314/x.png', 'notice') == "/images/notice/20260314/x.png"
+
+
+def test_non_temp_relative_path_kept_when_prefix_empty(monkeypatch):
+    monkeypatch.setattr(rich_text, 'FLASK_FILE_URL', '')
+    # 未配置文件域名时，相对路径同样原样保留
+    assert rich_text.save_file('/notice/20260314/x.png', 'notice') == "/notice/20260314/x.png"