DjangoAdmin_FastAPI_EleVue/tests/unit/modules/test_delete_cascade.py

"""删除级联防护：存在外键引用时禁止删除（dept→user、category→article、city→子级、dict/config→item）。"""
import json

import pytest

from core import R
from core.base_service import BaseService
from utils.request import parse_id_list

from modules.system.dept.repository import dept_repo
from modules.system.dept.service import delete_dept
from modules.system.role.models import Role, RoleMenu
from modules.system.user.repository import user_repo
from modules.category.repository import category_repo
from modules.category.service import delete_category
from modules.article.repository import article_repo
from modules.city.repository import city_repo
from modules.city.service import delete_city
from modules.dictionary.dict_item.repository import dictitem_repo
from modules.dictionary.dict.service import dict_service
from modules.configuration.config_item.repository import configitem_repo
from modules.configuration.config.service import config_service


class _FakeQuery:
    """模拟 SQLAlchemy Query：仅支持 first()，返回行或 None。"""

    def __init__(self, has_row):
        self._has_row = has_row

    def first(self):
        return object() if self._has_row else None


def _body(result):
    return json.loads(result.body)


def _allow_delete(monkeypatch):
    """放行批量删除，捕获传入的 ids 以便断言走到了该步骤。"""
    captured = {}

    def fake_batch_delete(repo, ids):
        captured['ids'] = ids
        return R.ok(msg="本次共删除{0}条数据".format(len(parse_id_list(str(ids)))))

    monkeypatch.setattr(BaseService, 'batch_delete_with_r', staticmethod(fake_batch_delete))
    return captured


# ============ dept → user ============

def test_delete_dept_blocks_when_users_reference(monkeypatch):
    monkeypatch.setattr(dept_repo, 'filter', lambda *a, **k: _FakeQuery(False))   # 无下级部门
    monkeypatch.setattr(user_repo, 'filter', lambda *a, **k: _FakeQuery(True))    # 有用户引用
    result = delete_dept('1')
    assert _body(result)['ok'] is False
    assert '用户' in _body(result)['msg']


def test_delete_dept_allows_when_no_references(monkeypatch):
    monkeypatch.setattr(dept_repo, 'filter', lambda *a, **k: _FakeQuery(False))
    monkeypatch.setattr(user_repo, 'filter', lambda *a, **k: _FakeQuery(False))
    captured = _allow_delete(monkeypatch)
    result = delete_dept('1,2')
    assert _body(result)['ok'] is True
    assert captured['ids'] == '1,2'


# ============ category → article ============

def test_delete_category_blocks_when_articles_reference(monkeypatch):
    monkeypatch.setattr(category_repo, 'filter', lambda *a, **k: _FakeQuery(False))   # 无下级分类
    monkeypatch.setattr(article_repo, 'filter', lambda *a, **k: _FakeQuery(True))     # 有文章引用
    result = delete_category('1')
    assert _body(result)['ok'] is False
    assert '文章' in _body(result)['msg']


def test_delete_category_allows_when_no_references(monkeypatch):
    monkeypatch.setattr(category_repo, 'filter', lambda *a, **k: _FakeQuery(False))
    monkeypatch.setattr(article_repo, 'filter', lambda *a, **k: _FakeQuery(False))
    captured = _allow_delete(monkeypatch)
    result = delete_category('1,2')
    assert _body(result)['ok'] is True
    assert captured['ids'] == '1,2'


# ============ city → 子级 ============

def test_delete_city_blocks_when_child_exists(monkeypatch):
    monkeypatch.setattr(city_repo, 'filter', lambda *a, **k: _FakeQuery(True))
    result = delete_city('1')
    assert _body(result)['ok'] is False
    assert '下级' in _body(result)['msg']


def test_delete_city_allows_when_no_children(monkeypatch):
    monkeypatch.setattr(city_repo, 'filter', lambda *a, **k: _FakeQuery(False))
    captured = _allow_delete(monkeypatch)
    result = delete_city('1,2')
    assert _body(result)['ok'] is True
    assert captured['ids'] == '1,2'


# ============ dict → dict_item ============

def test_dict_before_delete_blocks_when_items_exist(monkeypatch):
    monkeypatch.setattr(dictitem_repo, 'filter', lambda *a, **k: _FakeQuery(True))
    err = dict_service._before_delete('1,2')
    assert err is not None
    assert '字典项' in err


def test_dict_before_delete_allows_when_no_items(monkeypatch):
    monkeypatch.setattr(dictitem_repo, 'filter', lambda *a, **k: _FakeQuery(False))
    assert dict_service._before_delete('1,2') is None


# ============ config → config_item ============

def test_config_before_delete_blocks_when_items_exist(monkeypatch):
    monkeypatch.setattr(configitem_repo, 'filter', lambda *a, **k: _FakeQuery(True))
    err = config_service._before_delete('1,2')
    assert err is not None
    assert '配置项' in err


def test_config_before_delete_allows_when_no_items(monkeypatch):
    monkeypatch.setattr(configitem_repo, 'filter', lambda *a, **k: _FakeQuery(False))
    assert config_service._before_delete('1,2') is None


# ============ role → user_role / role_menu 残留清理 ============

@pytest.mark.asyncio
async def test_delete_role_cleans_user_role_and_role_menu(sqlite_db, monkeypatch):
    from modules.system.role import service as role_service
    from modules.system.role.models import Role
    from modules.system.role_menu.models import RoleMenu
    from modules.system.user_role.models import UserRole
    from core.database import get_db_session

    async def _noop(*args, **kwargs):
        return None

    monkeypatch.setattr(role_service, 'invalidate_role_perms', _noop)

    # 显式非内置 id（内置角色 1-4 禁止删除）
    role = Role(id=100, name='普通角色', code='R1', status=1, sort=1)
    role.save()
    db = get_db_session()
    db.add_all([
        UserRole(user_id=5, role_id=role.id),
        RoleMenu(role_id=role.id, menu_id=10),
    ])
    db.commit()

    resp = await role_service.delete_role(str(role.id))
    assert _body(resp)['code'] == 0
    # 关联残留已清理，角色本身软删
    assert db.query(UserRole).filter(UserRole.role_id == role.id).count() == 0
    assert db.query(RoleMenu).filter(RoleMenu.role_id == role.id).count() == 0
    assert role.is_delete == 1


# ============ position / level → user 引用（与部门一致） ============

def test_position_before_delete_blocks_when_users_reference(monkeypatch):
    from modules.system.position.service import position_service

    monkeypatch.setattr(user_repo, 'filter', lambda *a, **k: _FakeQuery(True))
    err = position_service._before_delete('1')
    assert err is not None
    assert '岗位' in err


def test_position_before_delete_allows_when_no_references(monkeypatch):
    from modules.system.position.service import position_service

    monkeypatch.setattr(user_repo, 'filter', lambda *a, **k: _FakeQuery(False))
    assert position_service._before_delete('1,2') is None


def test_level_before_delete_blocks_when_users_reference(monkeypatch):
    from modules.system.level.service import level_service

    monkeypatch.setattr(user_repo, 'filter', lambda *a, **k: _FakeQuery(True))
    err = level_service._before_delete('1')
    assert err is not None
    assert '职级' in err


def test_level_before_delete_allows_when_no_references(monkeypatch):
    from modules.system.level.service import level_service

    monkeypatch.setattr(user_repo, 'filter', lambda *a, **k: _FakeQuery(False))
    assert level_service._before_delete('1,2') is None