shanxia2.0/backend/tests/test_pl_dispatch_calendar_migration.py

"""用实际授权 INSERT 验证迁移：停用/隐藏入口不得派生维护权限。"""
from pathlib import Path
import re
import sqlite3
from uuid import uuid4

import pytest


MIGRATION = Path(__file__).resolve().parents[2] / 'docs/sql/migrations/2026-10-26-manual-dispatch-calendar.sql'


def database(prefix, *, existing_target=True):
    db = sqlite3.connect(':memory:')
    db.create_function('UUID', 0, lambda: str(uuid4()))
    db.create_function('NOW', 0, lambda: '2026-10-26 12:00:00')
    common = "tenant_no TEXT DEFAULT 't1', order_num INT, remark TEXT, create_by TEXT, create_dept_id TEXT, create_time TEXT, update_time TEXT, del_flag INT DEFAULT 0"
    db.execute(f"CREATE TABLE sys_menu (id TEXT PRIMARY KEY, menu_name TEXT, parent_id TEXT DEFAULT '0', menu_type TEXT, path TEXT, component TEXT, perms TEXT, is_frame INT, visible TEXT DEFAULT '1', status TEXT DEFAULT '1', {common})")
    db.execute(f"CREATE TABLE sys_role (id TEXT PRIMARY KEY, status TEXT DEFAULT '1', {common})")
    db.execute(f"CREATE TABLE sys_role_menu (id TEXT PRIMARY KEY, role_id TEXT, menu_id TEXT, {common})")
    def menu(record_id, parent_id, path=None, perms=None):
        db.execute('INSERT INTO sys_menu (id,parent_id,path,perms) VALUES (?,?,?,?)', (record_id, parent_id, path, perms))
    is_web = prefix == 'pl'
    menu('root', '0', '/web' if is_web else '/mini')
    menu('parent', 'root', '/pl' if is_web else '/mini/pl')
    if is_web:
        menu('folder', 'parent', 'calendar-module')
    menu('leaf', 'folder' if is_web else 'parent', perms=prefix + ':dispatch-calendar:list')
    menu('source', 'parent', perms='pl:plan:list' if is_web else 'mini:pl:plan:manage')
    if existing_target:
        menu('target', 'leaf', perms=prefix + ':dispatch-calendar:manage')
    db.execute("INSERT INTO sys_role (id) VALUES ('r1')")
    db.execute("INSERT INTO sys_role_menu (id,role_id,menu_id) VALUES ('calendar-grant','r1','leaf')")
    db.execute("INSERT INTO sys_role_menu (id,role_id,menu_id) VALUES ('source-grant','r1','source')")
    return db


def migrate(db):
    for sql in re.findall(r'INSERT INTO sys_(?:menu|role_menu)\s*\(.*?;', MIGRATION.read_text(), re.S):
        db.execute(sql)


def grants(db):
    return db.execute("SELECT COUNT(*) FROM sys_role_menu rm JOIN sys_menu m ON m.id=rm.menu_id WHERE m.perms LIKE '%dispatch-calendar:manage' AND rm.del_flag=0").fetchone()[0]


@pytest.mark.parametrize('prefix,part', [(prefix, part) for prefix in ['pl', 'mini:pl']
                         for part in ['source', 'leaf', 'parent', 'root', 'target']]
                         + [('pl', 'folder')])
@pytest.mark.parametrize('field', ['status', 'visible'])
def test_inactive_menu_cannot_grant_manage(prefix, part, field):
    with database(prefix) as db:
        db.execute(f'UPDATE sys_menu SET {field}=? WHERE id=?', ('0', part))
        migrate(db)
        assert grants(db) == 0


@pytest.mark.parametrize('prefix', ['pl', 'mini:pl'])
def test_active_permissions_are_granted_once_and_revocation_is_preserved(prefix):
    with database(prefix, existing_target=False) as db:
        migrate(db)
        migrate(db)
        assert grants(db) == 1
        db.execute("UPDATE sys_role_menu SET del_flag=1 WHERE menu_id IN (SELECT id FROM sys_menu WHERE perms LIKE '%dispatch-calendar:manage')")
        migrate(db)
        assert grants(db) == 0
        assert db.execute("SELECT COUNT(*) FROM sys_role_menu WHERE del_flag=1").fetchone()[0] == 1


@pytest.mark.parametrize('prefix', ['pl', 'mini:pl'])
@pytest.mark.parametrize('part', ['leaf', 'parent', 'root'])
def test_inactive_calendar_ancestors_do_not_create_manage_menu(prefix, part):
    with database(prefix, existing_target=False) as db:
        db.execute('UPDATE sys_menu SET status=? WHERE id=?', ('0', part))
        migrate(db)
        assert db.execute("SELECT COUNT(*) FROM sys_menu WHERE perms LIKE '%dispatch-calendar:manage'").fetchone()[0] == 0