shanxia2.0/backend/tests/test_system_child_diff.py

"""关联集合差量保存，仅使用独立内存 SQLite。"""
from datetime import datetime

import pytest
from sqlalchemy import create_engine, text
from sqlalchemy.orm import Session

from app.crud import user, role
from app.models.models import SysUserRole, SysUserPost, SysRole, SysRoleMenu, SysRoleDataScopeDetail, SysUser, SysPost, SysDept, SysMenu


@pytest.fixture
def db():
    engine = create_engine('sqlite:///:memory:')
    for model in (SysUserRole, SysUserPost, SysRole, SysRoleMenu, SysRoleDataScopeDetail, SysUser, SysPost, SysDept, SysMenu):
        model.__table__.create(engine)
    with engine.begin() as conn:
        conn.execute(text('CREATE UNIQUE INDEX uk_role_menu ON sys_role_menu(role_id, menu_id, del_flag)'))
    with Session(engine) as session:
        add(session, SysUser, 'u', user_no='u', nick_name='U')
        add(session, SysPost, 'p', post_name='post', post_code='P')
        add(session, SysMenu, 'm', menu_name='menu')
        add(session, SysDept, 'd', dept_name='dept')
        session.commit()
        yield session
    engine.dispose()


def add(db, model, id, **fields):
    row = model(id=id, tenant_no='t', del_flag=0, create_time=datetime(2020, 1, 1), **fields)
    db.add(row)
    db.flush()
    return row


def test_role_only_update_retains_posts_and_existing_identity(db):
    add(db, SysRole, 'r', role_name='role')
    add(db, SysRole, 'new', role_name='new')
    old = add(db, SysUserRole, 'ur', user_id='u', role_id='r')
    post = add(db, SysUserPost, 'up', user_id='u', post_id='p')
    user._replace_associations(db, 'u', 't', 'actor', role_ids=['r', 'new'])
    db.flush()
    assert old.del_flag == 0
    assert post.del_flag == 0
    assert old.create_time == datetime(2020, 1, 1)
    assert db.query(SysUserRole).filter_by(del_flag=0).count() == 2
    assert db.query(SysUserRole).count() == 2


def test_post_diff_clear_and_restore_retains_ids_and_tenant(db):
    old = add(db, SysUserPost, 'up', user_id='u', post_id='p')
    foreign = add(db, SysUserPost, 'foreign', user_id='u', post_id='foreign')
    foreign.tenant_no = 'other'
    db.flush()
    user.set_user_post_ids_only(db, 'u', 't', ['p'], 'actor')
    db.flush()
    assert old.del_flag == 0
    assert foreign.del_flag == 0
    user.set_user_post_ids_only(db, 'u', 't', [], 'actor')
    db.flush()
    assert old.del_flag == 1
    user.set_user_post_ids_only(db, 'u', 't', ['p', 'p'], 'actor')
    db.flush()
    assert old.del_flag == 0
    assert db.query(SysUserPost).count() == 2


def test_role_menus_noop_and_repeated_toggle_reuses_history(db):
    add(db, SysRole, 'r', role_name='role')
    old = add(db, SysRoleMenu, 'rm', role_id='r', menu_id='m')
    role.update_role_menus(db, 'r', ['m'], 't', 'actor')
    assert db.get(SysRoleMenu, 'rm') is not None
    assert old.del_flag == 0
    for _ in range(3):
        role.update_role_menus(db, 'r', [], 't', 'actor')
        assert old.del_flag == 1
        role.update_role_menus(db, 'r', ['m'], 't', 'actor')
        assert old.del_flag == 0
    assert db.query(SysRoleMenu).count() == 1


def test_role_data_scope_complete_save_omitted_clears_and_diff_reuses_id(db):
    add(db, SysRole, 'r', role_name='role', data_scope='5')
    old = add(db, SysRoleDataScopeDetail, 'scope', role_id='r', scope_type='dept', scope_value='d')
    role.update_role_data_scope(db, 'r', '5', None, 't', 'actor')
    assert old.del_flag == 1
    role.update_role_data_scope(db, 'r', '5', [{'scopeType': 'dept', 'scopeValue': 'd'}], 't', 'actor')
    assert old.del_flag == 0
    assert db.query(SysRoleDataScopeDetail).count() == 1
    role.update_role_data_scope(db, 'r', '5', [], 't', 'actor')
    assert old.del_flag == 1


def test_role_update_rejects_foreign_owner(db):
    owner = add(db, SysRole, 'r', role_name='role')
    owner.tenant_no = 'other'
    db.flush()
    with pytest.raises(ValueError):
        role.update_role_menus(db, 'r', ['m'], 't', 'actor')
    with pytest.raises(ValueError):
        role.update_role_data_scope(db, 'r', '5', [], 't', 'actor')


@pytest.mark.parametrize('target_model, fields, kwargs', [
    (SysRole, {'role_name': 'foreign'}, {'role_ids': ['foreign']}),
    (SysPost, {'post_name': 'foreign', 'post_code': 'F'}, {'post_ids': ['foreign']}),
])
def test_foreign_relation_target_rejected_before_changes(db, target_model, fields, kwargs):
    target = add(db, target_model, 'foreign', **fields)
    target.tenant_no = 'other'
    db.flush()
    with pytest.raises(ValueError):
        user._replace_associations(db, 'u', 't', 'actor', **kwargs)
    assert db.query(SysUserRole).count() == 0
    assert db.query(SysUserPost).count() == 0


def test_foreign_scope_target_rejected(db):
    add(db, SysRole, 'r', role_name='role')
    target = db.get(SysDept, 'd')
    target.tenant_no = 'other'
    db.flush()
    with pytest.raises(ValueError):
        role.update_role_data_scope(db, 'r', '5', [{'scopeType':'dept','scopeValue':'d'}], 't', 'actor')


def test_create_user_does_not_commit_parent_when_associations_fail(db, monkeypatch):
    monkeypatch.setattr(user, '_hash_password', lambda value: 'hashed')
    def fail(*args, **kwargs):
        raise ValueError('invalid association')
    monkeypatch.setattr(user, '_replace_associations', fail)
    with pytest.raises(ValueError):
        user.create_user(db, user_no='new-u', nick_name='new', password='password', tenant_no='t')
    db.rollback()
    assert db.query(SysUser).filter_by(user_no='new-u').first() is None


def test_user_update_rejects_foreign_owner_before_header_change(db):
    with pytest.raises(ValueError):
        user.update_user(db, 'u', tenant_no='other', nick_name='hijacked', role_ids=[])
    assert db.get(SysUser, 'u').nick_name == 'U'


def test_role_menu_rejects_foreign_target(db):
    add(db, SysRole, 'r', role_name='role')
    db.get(SysMenu, 'm').tenant_no = 'other'
    db.flush()
    with pytest.raises(ValueError):
        role.update_role_menus(db, 'r', ['m'], 't', 'actor')
    assert db.query(SysRoleMenu).count() == 0