zkxw/backend/tests/test_market_mobile_permissions.py

import asyncio

import pytest
from fastapi import HTTPException
from sqlalchemy import create_engine
from sqlalchemy.orm import sessionmaker

from app.core.mobile_menu_guard import (
    require_market_menu_code,
    require_market_menu_permission,
    user_has_tenant_menu_perm,
)
from app.models.base import Base
from app.models.models import (
    BizPresignOrder,
    BizPresignOrderSampleType,
    BizPresignTransfer,
    SysMenu,
    SysDept,
    SysPost,
    SysPostRole,
    SysRole,
    SysRoleDataScopeDetail,
    SysRoleMenu,
    SysUser,
    SysUserPost,
    SysUserRole,
)


def _session():
    engine = create_engine("sqlite:///:memory:")
    Base.metadata.create_all(
        engine,
        tables=[
            SysMenu.__table__,
            SysDept.__table__,
            SysPost.__table__,
            SysPostRole.__table__,
            SysRole.__table__,
            SysRoleDataScopeDetail.__table__,
            SysRoleMenu.__table__,
            SysUser.__table__,
            SysUserPost.__table__,
            SysUserRole.__table__,
            BizPresignOrder.__table__,
            BizPresignOrderSampleType.__table__,
            BizPresignTransfer.__table__,
        ],
    )
    return sessionmaker(bind=engine)()


def _add_role_menu(
    db,
    *,
    tenant_no,
    role_id,
    menu_id,
    perm_code,
    role_name="市场角色",
    data_scope="1",
):
    db.add(
        SysRole(
            id=role_id,
            role_name=role_name,
            data_scope=data_scope,
            status="1",
            tenant_no=tenant_no,
            create_dept_id="",
            del_flag=0,
        )
    )
    db.add(
        SysMenu(
            id=menu_id,
            menu_name=perm_code,
            parent_id="root",
            menu_type="C",
            path=f"/{menu_id}",
            perms=perm_code,
            visible="1",
            status="1",
            tenant_no=tenant_no,
            create_dept_id="",
            del_flag=0,
        )
    )
    db.add(
        SysRoleMenu(
            id=f"rm-{role_id}-{menu_id}",
            role_id=role_id,
            menu_id=menu_id,
            tenant_no=tenant_no,
            create_dept_id="",
            del_flag=0,
        )
    )


def test_specialist_permission_does_not_grant_manager_access():
    assert not require_market_menu_permission(
        user_id="u1",
        tenant_no="t1",
        required="mini:market:manager:presign",
        granted={"mini:market:specialist:presign"},
    )


def test_tenant_menu_permission_does_not_fall_back_to_default_tenant():
    db = _session()
    _add_role_menu(
        db,
        tenant_no="tenant-a",
        role_id="role-a",
        menu_id="menu-a",
        perm_code="mini:market:manager:presign",
    )
    _add_role_menu(
        db,
        tenant_no="tenant-b",
        role_id="role-b",
        menu_id="menu-b",
        perm_code="mini:market:specialist:presign",
    )
    db.add_all(
        [
            SysUserRole(
                id="ur-a",
                user_id="u1",
                role_id="role-a",
                tenant_no="tenant-a",
                create_dept_id="",
                del_flag=0,
            ),
            SysUserRole(
                id="ur-b",
                user_id="u1",
                role_id="role-b",
                tenant_no="tenant-b",
                create_dept_id="",
                del_flag=0,
            ),
        ]
    )
    db.commit()

    assert user_has_tenant_menu_perm(
        db,
        user_id="u1",
        tenant_no="tenant-b",
        perm_code="mini:market:specialist:presign",
    )
    assert not user_has_tenant_menu_perm(
        db,
        user_id="u1",
        tenant_no="tenant-b",
        perm_code="mini:market:manager:presign",
    )


def test_tenant_menu_permission_includes_post_inherited_roles():
    db = _session()
    _add_role_menu(
        db,
        tenant_no="tenant-b",
        role_id="role-b",
        menu_id="menu-b",
        perm_code="mini:market:specialist:presign",
    )
    db.add_all(
        [
            SysPost(
                id="post-b",
                post_name="市场专员",
                post_code="specialist",
                status="1",
                tenant_no="tenant-b",
                create_dept_id="",
                del_flag=0,
            ),
            SysUserPost(
                id="up-b",
                user_id="u1",
                post_id="post-b",
                tenant_no="tenant-b",
                create_dept_id="",
                del_flag=0,
            ),
            SysPostRole(
                id="pr-b",
                post_id="post-b",
                role_id="role-b",
                tenant_no="tenant-b",
                create_dept_id="",
                del_flag=0,
            ),
        ]
    )
    db.commit()

    assert user_has_tenant_menu_perm(
        db,
        user_id="u1",
        tenant_no="tenant-b",
        perm_code="mini:market:specialist:presign",
    )


def test_strict_guard_rejects_missing_tenant_identity():
    dependency = require_market_menu_code("mini:market:specialist:presign")

    with pytest.raises(HTTPException) as exc_info:
        asyncio.run(dependency(user={"user_id": "u1", "tenant_no": ""}, db=None))

    assert exc_info.value.status_code == 403
    assert exc_info.value.detail == "用户租户或身份信息缺失"


def test_strict_guard_does_not_trust_cross_tenant_super_admin_role():
    db = _session()
    db.add_all(
        [
            SysRole(
                id="tenant-b-role",
                role_name="市场专员",
                status="1",
                tenant_no="tenant-b",
                create_dept_id="",
                del_flag=0,
            ),
            SysUserRole(
                id="tenant-b-user-role",
                user_id="u1",
                role_id="tenant-b-role",
                tenant_no="tenant-b",
                create_dept_id="",
                del_flag=0,
            ),
        ]
    )
    db.commit()
    dependency = require_market_menu_code("mini:market:specialist:presign")

    with pytest.raises(HTTPException) as exc_info:
        asyncio.run(
            dependency(
                user={
                    "user_id": "u1",
                    "tenant_no": "tenant-b",
                    "role": "super_admin",
                },
                db=db,
            )
        )

    assert exc_info.value.status_code == 403
    assert exc_info.value.detail == "无移动端菜单权限"


def test_strict_guard_uses_current_tenant_super_admin_role():
    db = _session()
    db.add_all(
        [
            SysRole(
                id="tenant-b-super-role",
                role_name="超级管理员",
                status="1",
                tenant_no="tenant-b",
                create_dept_id="",
                del_flag=0,
            ),
            SysUserRole(
                id="tenant-b-super-user-role",
                user_id="u1",
                role_id="tenant-b-super-role",
                tenant_no="tenant-b",
                create_dept_id="",
                del_flag=0,
            ),
        ]
    )
    db.commit()
    dependency = require_market_menu_code("mini:market:specialist:presign")
    user = {"user_id": "u1", "tenant_no": "tenant-b", "role": "admin"}

    result = asyncio.run(dependency(user=user, db=db))

    assert result is user


def _add_auth_user(db, *, tenant_no="tenant-b"):
    db.add(
        SysUser(
            id="u1",
            user_no="u1",
            nick_name="租户用户",
            password="unused",
            status="1",
            tenant_no=tenant_no,
            create_dept_id="dept-b",
            del_flag=0,
        )
    )


def _add_cross_tenant_roles(db):
    from app.core.config import settings

    default_tenant = settings.DEFAULT_TENANT_NO
    _add_role_menu(
        db,
        tenant_no=default_tenant,
        role_id="default-super",
        menu_id="default-menu",
        perm_code="default:anything",
        role_name="超级管理员",
        data_scope="1",
    )
    _add_role_menu(
        db,
        tenant_no="tenant-b",
        role_id="tenant-b-specialist",
        menu_id="tenant-b-presign",
        perm_code="mini:market:specialist:presign",
        role_name="市场专员",
        data_scope="4",
    )
    db.add_all(
        [
            SysUserRole(
                id="default-user-role",
                user_id="u1",
                role_id="default-super",
                tenant_no=default_tenant,
                create_dept_id="",
                del_flag=0,
            ),
            SysUserRole(
                id="tenant-b-user-role",
                user_id="u1",
                role_id="tenant-b-specialist",
                tenant_no="tenant-b",
                create_dept_id="",
                del_flag=0,
            ),
        ]
    )


def _add_presign(db, order_id, *, create_by):
    db.add(
        BizPresignOrder(
            id=order_id,
            mother_name=order_id,
            sales_user_id="u1",
            order_status="active",
            tenant_no="tenant-b",
            create_by=create_by,
            create_dept_id="dept-b",
            del_flag=0,
        )
    )


def _authenticate(db, monkeypatch):
    from app.core import auth

    monkeypatch.setattr(
        auth,
        "decode_token",
        lambda token: {"sub": "u1", "sid": "session-u1"},
    )

    async def valid_session(session_id):
        return session_id == "session-u1"

    async def extend_session(session_id):
        return None

    monkeypatch.setattr(auth, "is_session_valid", valid_session)
    monkeypatch.setattr(auth, "extend_session", extend_session)
    return asyncio.run(auth.get_current_user_by_token("token", db))


def test_authenticated_tenant_role_drives_menu_and_presign_data_scope(
    monkeypatch,
):
    from app.services.market_field_scope import MarketFieldContext
    from app.services.market_presign import list_presign_orders

    db = _session()
    _add_auth_user(db)
    _add_cross_tenant_roles(db)
    _add_presign(db, "own", create_by="u1")
    _add_presign(db, "field-only", create_by="other")
    db.commit()

    current_user = _authenticate(db, monkeypatch)
    dependency = require_market_menu_code("mini:market:specialist:presign")
    guarded_user = asyncio.run(dependency(user=current_user, db=db))
    rows, total = list_presign_orders(
        db,
        ctx=MarketFieldContext("tenant-b", "u1", "specialist"),
        user_info=guarded_user,
    )

    assert current_user["tenant_no"] == "tenant-b"
    assert current_user["role"] == "admin"
    assert current_user["role_names"] == ["市场专员"]
    assert total == 1
    assert [row["id"] for row in rows] == ["own"]


def test_authenticated_current_tenant_super_admin_remains_unrestricted(
    monkeypatch,
):
    from app.services.market_field_scope import MarketFieldContext
    from app.services.market_presign import list_presign_orders

    db = _session()
    _add_auth_user(db)
    db.add_all(
        [
            SysRole(
                id="tenant-b-super",
                role_name="超级管理员",
                data_scope="1",
                status="1",
                tenant_no="tenant-b",
                create_dept_id="",
                del_flag=0,
            ),
            SysUserRole(
                id="tenant-b-super-user-role",
                user_id="u1",
                role_id="tenant-b-super",
                tenant_no="tenant-b",
                create_dept_id="",
                del_flag=0,
            ),
        ]
    )
    _add_presign(db, "own", create_by="u1")
    _add_presign(db, "field-only", create_by="other")
    db.commit()

    current_user = _authenticate(db, monkeypatch)
    dependency = require_market_menu_code("mini:market:specialist:presign")
    guarded_user = asyncio.run(dependency(user=current_user, db=db))
    rows, total = list_presign_orders(
        db,
        ctx=MarketFieldContext("tenant-b", "u1", "specialist"),
        user_info=guarded_user,
    )

    assert current_user["role"] == "super_admin"
    assert current_user["role_names"] == ["超级管理员"]
    assert total == 2
    assert {row["id"] for row in rows} == {"own", "field-only"}


def test_token_issuance_uses_the_users_current_tenant_roles(monkeypatch):
    from app.services import issue_user_tokens

    db = _session()
    _add_auth_user(db)
    _add_cross_tenant_roles(db)
    db.commit()

    async def no_op(**kwargs):
        return None

    monkeypatch.setattr(issue_user_tokens, "store_session", no_op)
    monkeypatch.setattr(issue_user_tokens, "store_refresh_token", no_op)

    issued = asyncio.run(
        issue_user_tokens.issue_tokens_for_user(
            db, db.query(SysUser).filter_by(id="u1").one()
        )
    )

    assert issued["userInfo"]["role"] == "admin"
    assert issued["userInfo"]["roleNames"] == ["市场专员"]


def test_public_auth_entrypoints_pass_the_users_tenant_to_role_queries(
    monkeypatch,
):
    from app.api import public
    from app.schemas.auth import ProfileUpdateRequest

    db = _session()
    _add_auth_user(db)
    db.commit()
    user = db.query(SysUser).filter_by(id="u1").one()
    calls = {"active": [], "role": [], "names": [], "menus": []}

    def has_active_roles(db_arg, user_id, tenant_no=None):
        calls["active"].append((user_id, tenant_no))
        return True

    def resolve_role(db_arg, user_id, tenant_no=None):
        calls["role"].append((user_id, tenant_no))
        return "admin"

    def role_names(db_arg, user_id, tenant_no=None):
        calls["names"].append((user_id, tenant_no))
        return ["市场专员"]

    def menu_codes(db_arg, user_id, tenant_no=None):
        calls["menus"].append((user_id, tenant_no))
        return {"mini:market:specialist:presign"}

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

    async def valid(*args, **kwargs):
        return True

    current_user = {
        "user_id": "u1",
        "tenant_no": "tenant-b",
        "role": "admin",
    }

    monkeypatch.setattr(public.login_crypto, "ensure_initialized", lambda: None)
    monkeypatch.setattr(
        public.login_crypto,
        "extract_login_credentials",
        lambda body: ("u1", "password", None),
    )
    monkeypatch.setattr(public.user_crud, "get_by_user_no", lambda db_arg, no: user)
    monkeypatch.setattr(public, "verify_password", lambda raw, hashed: True)
    monkeypatch.setattr(public, "user_has_active_roles", has_active_roles)
    monkeypatch.setattr(public, "resolve_web_role_for_user", resolve_role)
    monkeypatch.setattr(public, "get_user_role_names", role_names)
    monkeypatch.setattr(public, "iter_user_menu_perm_codes", menu_codes)
    monkeypatch.setattr(public, "create_access_token", lambda **kwargs: "access")
    monkeypatch.setattr(public, "create_refresh_token", lambda **kwargs: "refresh")
    monkeypatch.setattr(public, "store_session", no_op)
    monkeypatch.setattr(public, "store_refresh_token", no_op)
    monkeypatch.setattr(public, "revoke_refresh_token", no_op)
    monkeypatch.setattr(public, "extend_session", no_op)
    monkeypatch.setattr(public, "is_session_valid", valid)
    monkeypatch.setattr(public, "is_refresh_token_valid", valid)
    monkeypatch.setattr(
        public,
        "decode_token",
        lambda token: {
            "type": "refresh",
            "sid": "session-u1",
            "jti": "refresh-u1",
            "sub": "u1",
        },
    )
    monkeypatch.setattr(public, "update_user", lambda *args, **kwargs: None)

    async def resolve_current(request, db_arg):
        return current_user

    async def current_by_token(token, db):
        return current_user

    monkeypatch.setattr(public, "_resolve_current_user_from_request", resolve_current)
    monkeypatch.setattr(public, "get_current_user_by_token", current_by_token)

    class RequestStub:
        headers = {"Authorization": "Bearer token"}

    asyncio.run(public.web_login(request=None, body={}, db=db))
    asyncio.run(public.refresh_token({"refreshToken": "refresh"}, db=db))
    asyncio.run(
        public.update_current_user_profile(
            request=RequestStub(),
            body=ProfileUpdateRequest(),
            db=db,
        )
    )
    asyncio.run(public.get_web_auth_menu_codes(request=RequestStub(), db=db))

    assert calls == {
        "active": [("u1", "tenant-b")],
        "role": [("u1", "tenant-b")] * 3,
        "names": [("u1", "tenant-b")] * 2,
        "menus": [("u1", "tenant-b")],
    }


def test_org_labels_are_tenant_scoped_in_auth_profile_and_token_payloads(
    monkeypatch,
):
    from app.api import public
    from app.core.config import settings
    from app.crud.user import list_post_ids_for_user
    from app.schemas.auth import ProfileUpdateRequest
    from app.services import issue_user_tokens

    db = _session()
    _add_auth_user(db)
    _add_role_menu(
        db,
        tenant_no="tenant-b",
        role_id="tenant-b-role",
        menu_id="tenant-b-menu",
        perm_code="mini:market:specialist:presign",
        role_name="市场专员",
        data_scope="4",
    )
    user = db.query(SysUser).filter_by(id="u1").one()
    user.dept_id = "default-dept"
    db.add_all(
        [
            SysUserRole(
                id="tenant-b-role-link",
                user_id="u1",
                role_id="tenant-b-role",
                tenant_no="tenant-b",
                create_dept_id="",
                del_flag=0,
            ),
            SysDept(
                id="default-dept",
                dept_name="默认租户部门",
                tenant_no=settings.DEFAULT_TENANT_NO,
                create_dept_id="",
                del_flag=0,
            ),
            SysDept(
                id="tenant-b-dept",
                dept_name="租户B部门",
                tenant_no="tenant-b",
                create_dept_id="",
                del_flag=0,
            ),
            SysDept(
                id="tenant-b-deleted-dept",
                dept_name="租户B已删除部门",
                tenant_no="tenant-b",
                create_dept_id="",
                del_flag=1,
            ),
            SysPost(
                id="default-post",
                post_name="默认租户职位",
                post_code="default-post",
                status="1",
                tenant_no=settings.DEFAULT_TENANT_NO,
                create_dept_id="",
                del_flag=0,
            ),
            SysPost(
                id="tenant-b-post",
                post_name="租户B有效职位",
                post_code="tenant-b-post",
                status="1",
                tenant_no="tenant-b",
                create_dept_id="",
                del_flag=0,
            ),
            SysPost(
                id="tenant-b-disabled-post",
                post_name="租户B停用职位",
                post_code="tenant-b-disabled-post",
                status="0",
                tenant_no="tenant-b",
                create_dept_id="",
                del_flag=0,
            ),
            SysPost(
                id="tenant-b-deleted-post",
                post_name="租户B已删除职位",
                post_code="tenant-b-deleted-post",
                status="1",
                tenant_no="tenant-b",
                create_dept_id="",
                del_flag=1,
            ),
            SysPost(
                id="tenant-b-unbound-post",
                post_name="租户B已解绑职位",
                post_code="tenant-b-unbound-post",
                status="1",
                tenant_no="tenant-b",
                create_dept_id="",
                del_flag=0,
            ),
            SysUserPost(
                id="default-user-post",
                user_id="u1",
                post_id="default-post",
                tenant_no=settings.DEFAULT_TENANT_NO,
                create_dept_id="",
                del_flag=0,
            ),
            SysUserPost(
                id="tenant-b-user-post",
                user_id="u1",
                post_id="tenant-b-post",
                tenant_no="tenant-b",
                create_dept_id="",
                del_flag=0,
            ),
            SysUserPost(
                id="tenant-b-disabled-user-post",
                user_id="u1",
                post_id="tenant-b-disabled-post",
                tenant_no="tenant-b",
                create_dept_id="",
                del_flag=0,
            ),
            SysUserPost(
                id="tenant-b-deleted-user-post",
                user_id="u1",
                post_id="tenant-b-deleted-post",
                tenant_no="tenant-b",
                create_dept_id="",
                del_flag=0,
            ),
            SysUserPost(
                id="tenant-b-unbound-user-post",
                user_id="u1",
                post_id="tenant-b-unbound-post",
                tenant_no="tenant-b",
                create_dept_id="",
                del_flag=1,
            ),
        ]
    )
    db.commit()

    dept_name, post_names = issue_user_tokens.resolve_user_org_labels(db, user)
    assert dept_name is None
    assert post_names == ["租户B有效职位"]
    assert set(list_post_ids_for_user(db, "u1")) == {
        "default-post",
        "tenant-b-post",
        "tenant-b-disabled-post",
        "tenant-b-deleted-post",
    }
    assert set(list_post_ids_for_user(db, "u1", tenant_no="tenant-b")) == {
        "tenant-b-post",
        "tenant-b-disabled-post",
        "tenant-b-deleted-post",
    }

    user.dept_id = "tenant-b-deleted-dept"
    db.commit()
    dept_name, post_names = issue_user_tokens.resolve_user_org_labels(db, user)
    assert dept_name is None
    assert post_names == ["租户B有效职位"]

    user.dept_id = "tenant-b-dept"
    db.commit()
    authenticated = _authenticate(db, monkeypatch)
    assert authenticated["dept_name"] == "租户B部门"
    assert authenticated["post_names"] == ["租户B有效职位"]

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

    monkeypatch.setattr(issue_user_tokens, "store_session", no_op)
    monkeypatch.setattr(issue_user_tokens, "store_refresh_token", no_op)
    issued = asyncio.run(issue_user_tokens.issue_tokens_for_user(db, user))
    assert issued["userInfo"]["deptName"] == "租户B部门"
    assert issued["userInfo"]["postNames"] == ["租户B有效职位"]

    current_user = {"user_id": "u1", "tenant_no": "tenant-b"}

    async def resolve_current(request, db_arg):
        return current_user

    monkeypatch.setattr(public, "_resolve_current_user_from_request", resolve_current)
    monkeypatch.setattr(public, "update_user", lambda *args, **kwargs: None)
    profiled = asyncio.run(
        public.update_current_user_profile(
            request=None,
            body=ProfileUpdateRequest(),
            db=db,
        )
    )
    assert profiled["data"]["deptName"] == "租户B部门"
    assert profiled["data"]["postNames"] == ["租户B有效职位"]