shanxia2.0/backend/tests/test_role_workflow_semantics_restoration.py

"""Restore actual full-save/metadata semantics; UI handler and SQLite persistence."""
import json
from datetime import date, datetime
from pathlib import Path
import subprocess

import pymysql
import pytest
from sqlalchemy import create_engine
from sqlalchemy.orm import Session

from app.models.models import SysRole, SysRoleDataScopeDetail, SysDept, SysUser
from app.models.bx_models import BxWorkflowDef, BxWorkflowNode, BxReimburseApply, BxAdvanceApply
from app.core.bx_auth import BxUserContext
from app.services.bx.workflow import WorkflowService

USER = BxUserContext('u', 'User', 'd', 'super_admin', 't')


@pytest.fixture
def db(monkeypatch):
    def forbidden(*args, **kwargs):
        raise AssertionError('Business MySQL connections are forbidden')
    monkeypatch.setattr(pymysql, 'connect', forbidden)
    monkeypatch.setattr(pymysql.connections.Connection, 'connect', forbidden)
    engine = create_engine('sqlite://')
    for model in (SysRole, SysRoleDataScopeDetail, SysDept, SysUser,
                  BxWorkflowDef, BxWorkflowNode, BxReimburseApply, BxAdvanceApply):
        model.__table__.create(engine)
    with Session(engine, autoflush=False) as session:
        yield session
    engine.dispose()


def actual_scope_payload(departments, users=()):
    root = Path(__file__).parents[2]
    program = r'''
const fs = require('node:fs')
const ts = require(process.argv[1])
const source = fs.readFileSync(process.argv[2], 'utf8').match(/<script[^>]*>([\s\S]*?)<\/script>/)[1]
const ast = ts.createSourceFile('dialog.ts', source, ts.ScriptTarget.Latest, true)
const fn = ast.statements.find(n => ts.isFunctionDeclaration(n) && n.name.text === 'handleDataScopeSubmit')
if (!fn) throw Error('handler missing')
const js = ts.transpileModule(fn.getText(ast), {compilerOptions:{target:ts.ScriptTarget.ES2022}}).outputText
let payload
const selections = JSON.parse(process.argv[3])
const context = {
 dataScopeSubmitLoading: {value:false}, dataScopeForm:{dataScope:'5'},
 deptTreeRef:{value:{getCheckedKeys:()=>selections.departments}},
 scopeUserCheckedKeys:{value:selections.users}, dataScopeEditingRoleId:{value:'role'},
 fetchUpdateRoleDataScope:async (_id, data)=>{payload=data},
 dataScopeDialogVisible:{value:true}, emit:()=>{}, refreshViewIfOpen:async()=>{}
}
const run = new Function(...Object.keys(context), js+';return handleDataScopeSubmit')(...Object.values(context))
run().then(()=>process.stdout.write(JSON.stringify(payload))).catch(e=>{console.error(e);process.exit(1)})
'''
    result = subprocess.run(['node', '-e', program, str(root/'web-admin/node_modules/typescript'),
        str(root/'web-admin/src/components/business/role-manage-dialogs/index.vue'),
        json.dumps({'departments': departments, 'users': list(users)})],
        check=True, text=True, capture_output=True)
    return json.loads(result.stdout)


def role_scope(db):
    db.add(SysRole(id='role', tenant_no='t', role_name='Role', data_scope='5'))
    db.add(SysDept(id='dept', tenant_no='t', dept_name='Department'))
    old = SysRoleDataScopeDetail(id='scope', tenant_no='t', role_id='role', scope_type='dept',
        scope_value='dept', create_by='creator', create_time=datetime(2020, 1, 1))
    foreign = SysRoleDataScopeDetail(id='foreign', tenant_no='other', role_id='role', scope_type='dept', scope_value='other-dept')
    db.add_all([old, foreign]); db.commit()
    return old, foreign


def scope_route(db, monkeypatch, payload):
    from app.api import admin
    monkeypatch.setattr(admin, '_get_tenant_no', lambda *args: 't')
    return admin.update_role_data_scope('role', admin.RoleDataScopeBody.model_validate(payload),
                                       user={'user_id':'u'}, db=db)


def test_actual_ui_clear_to_schema_route_removes_custom_scope(db, monkeypatch):
    old, foreign = role_scope(db)
    payload = actual_scope_payload([])
    assert payload == {'dataScope': '5'}  # The current complete-save UI intentionally omits an empty list.
    result = scope_route(db, monkeypatch, payload)
    db.refresh(old); db.refresh(foreign)
    assert result['code'] == 200 and old.del_flag == 1 and foreign.del_flag == 0


def test_actual_ui_unchanged_and_clear_restore_keep_relation_identity(db, monkeypatch):
    old, foreign = role_scope(db)
    for selected in (['dept'], [], ['dept'], ['dept']):
        result = scope_route(db, monkeypatch, actual_scope_payload(selected))
        assert result['code'] == 200
        db.refresh(old)
        assert old.del_flag == (0 if selected else 1)
        assert old.create_by == 'creator' and old.create_time == datetime(2020, 1, 1)
    assert db.query(SysRoleDataScopeDetail).filter_by(tenant_no='t').count() == 1
    assert foreign.del_flag == 0


def test_explicit_empty_and_noncustom_scope_clear_still_supported(db, monkeypatch):
    old, _ = role_scope(db)
    assert scope_route(db, monkeypatch, {'dataScope':'5', 'scopeDetails':[]})['code'] == 200
    assert old.del_flag == 1
    assert scope_route(db, monkeypatch, {'dataScope':'1'})['code'] == 200
    assert old.del_flag == 1


def test_foreign_scope_target_is_rejected_before_mutation(db, monkeypatch):
    old, foreign = role_scope(db)
    db.add(SysDept(id='foreign-dept', tenant_no='other', dept_name='Foreign')); db.commit()
    response = scope_route(db, monkeypatch, {'dataScope':'5', 'scopeDetails':[{'scopeType':'dept','scopeValue':'foreign-dept'}]})
    assert response['code'] == 400 and old.del_flag == foreign.del_flag == 0
    assert not db.dirty and not db.new


def pending_workflow(db):
    workflow = BxWorkflowDef(id='wf', tenant_no='t', biz_type='reimburse', wf_name='Workflow', status=1)
    node = BxWorkflowNode(id='node', tenant_no='t', wf_id='wf', node_name='Review', node_order=1,
        action_type='approve', approver_id='u', approver_name='User', create_time=datetime(2020, 1, 1))
    application = BxReimburseApply(id='a', tenant_no='t', apply_code='A', title='Apply', applicant_id='u',
        applicant_name='User', expense_id='expense', expense_name='Expense', amount=10,
        reimburse_date=date(2026, 1, 1), status='pending', wf_id='wf', current_node='Review')
    db.add_all([workflow, node, application]); db.commit()
    return workflow, node, application


def test_pending_workflow_name_update_through_real_schema_route_keeps_nodes(db, monkeypatch):
    from app.api.bx import workflow as api
    from app.schemas.bx.workflow import WorkflowDefUpdate
    workflow, node, application = pending_workflow(db)
    monkeypatch.setattr(api, 'get_tenant_no', lambda *args: 't')
    api.update_workflow('wf', WorkflowDefUpdate(wfName='Renamed'), db=db, user=USER)
    assert workflow.wf_name == 'Renamed'
    assert node.id == 'node' and node.node_name == 'Review' and node.create_time == datetime(2020, 1, 1)
    assert application.status == 'pending' and application.current_node == 'Review'
    assert db.query(BxWorkflowNode).count() == 1


def test_pending_workflow_remark_and_unchanged_status_are_metadata_only(db):
    workflow, node, application = pending_workflow(db)
    result = WorkflowService.update(db, 'wf', {'remark':'Note', 'status':1, 'biz_type':'reimburse'}, user=USER)
    assert result.remark == 'Note' and node.node_name == 'Review' and application.status == 'pending'


@pytest.mark.parametrize('patch', [{'wf_name':'Rename', 'status':0}, {'wf_name':'Rename', 'biz_type':'advance'}])
def test_pending_structural_header_changes_still_rejected_atomically(db, patch):
    workflow, node, application = pending_workflow(db)
    with pytest.raises(ValueError, match='审批'):
        WorkflowService.update(db, 'wf', patch, user=USER)
    assert workflow.wf_name == 'Workflow' and workflow.biz_type == 'reimburse' and workflow.status == 1
    assert not db.dirty and not db.new


def test_pending_node_change_and_delete_guards_remain(db):
    workflow, node, application = pending_workflow(db)
    with pytest.raises(ValueError, match='审批'):
        WorkflowService.save_nodes(db, 'wf', [{'id':'node', 'node_name':'Changed', 'node_order':1}], user=USER)
    with pytest.raises(ValueError, match='审批'):
        WorkflowService.soft_delete(db, 'wf', user=USER)
    assert node.node_name == 'Review' and workflow.del_flag == 0 and not db.dirty and not db.new