zkxw/backend/tests/test_market_field_scope.py

import re
from pathlib import Path

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

from app.models.base import Base
from app.models.models import (
    BizCollectionProtocol,
    BizCollectionRegionStaff,
    BizPresignOrder,
    BizProtocolCollectionInfo,
    SysPost,
    SysPostRole,
    SysRole,
    SysRoleDataScopeDetail,
    SysUserPost,
    SysUserRole,
)
from app.models.research_donation import (
    BizResearchDonation,
    BizResearchDonationCollection,
)


def _session():
    engine = create_engine("sqlite:///:memory:")
    Base.metadata.create_all(
        engine,
        tables=[
            BizCollectionRegionStaff.__table__,
            BizPresignOrder.__table__,
            BizCollectionProtocol.__table__,
            BizProtocolCollectionInfo.__table__,
            BizResearchDonation.__table__,
            BizResearchDonationCollection.__table__,
            SysRole.__table__,
            SysRoleDataScopeDetail.__table__,
            SysUserRole.__table__,
            SysPost.__table__,
            SysPostRole.__table__,
            SysUserPost.__table__,
        ],
    )
    return sessionmaker(bind=engine)()


def _presign(
    order_id,
    *,
    tenant_no="tenant-a",
    region_id="region-a",
    sales_user_id="sales-a",
    manager_user_id="manager-a",
    create_by="creator-a",
    del_flag=0,
):
    return BizPresignOrder(
        id=order_id,
        tenant_no=tenant_no,
        region_id=region_id,
        sales_user_id=sales_user_id,
        region_manager_user_id=manager_user_id,
        create_by=create_by,
        create_dept_id="dept-a",
        del_flag=del_flag,
    )


def _cell_protocol(
    protocol_id,
    *,
    tenant_no="tenant-a",
    collection_tenant_no=None,
    region_id="region-a",
    sales_user_id="sales-a",
    manager_user_id="manager-a",
    parent_del_flag=0,
    collection_del_flag=0,
):
    collection_tenant_no = collection_tenant_no or tenant_no
    return [
        BizCollectionProtocol(
            id=protocol_id,
            protocol_no=f"CP-{protocol_id}",
            tenant_no=tenant_no,
            create_by=sales_user_id,
            create_dept_id="dept-a",
            del_flag=parent_del_flag,
        ),
        BizProtocolCollectionInfo(
            id=f"ci-{protocol_id}",
            collection_protocol_id=protocol_id,
            tenant_no=collection_tenant_no,
            region_id=region_id,
            sales_user_id=sales_user_id,
            region_manager_user_id=manager_user_id,
            create_dept_id="dept-a",
            del_flag=collection_del_flag,
        ),
    ]


def _donation(
    donation_id,
    *,
    tenant_no="tenant-a",
    collection_tenant_no=None,
    region_id="region-a",
    sales_user_id="sales-a",
    manager_user_id="manager-a",
    parent_del_flag=0,
    collection_del_flag=0,
):
    collection_tenant_no = collection_tenant_no or tenant_no
    return [
        BizResearchDonation(
            id=donation_id,
            donation_no=f"RD-{donation_id}",
            tenant_no=tenant_no,
            create_by=sales_user_id,
            create_dept_id="dept-a",
            del_flag=parent_del_flag,
        ),
        BizResearchDonationCollection(
            id=f"dc-{donation_id}",
            research_donation_id=donation_id,
            tenant_no=collection_tenant_no,
            region_id=region_id,
            sales_user_id=sales_user_id,
            region_manager_user_id=manager_user_id,
            create_dept_id="dept-a",
            del_flag=collection_del_flag,
        ),
    ]


def _ids(rows):
    return {row.id for row in rows}


def _user_info(user_id="u1", role="super_admin"):
    return {"user_id": user_id, "role": role, "dept_id": "dept-a"}


def test_specialist_matches_sales_user_only():
    from app.services.market_field_scope import (
        MarketFieldContext,
        apply_presign_scope,
    )

    db = _session()
    db.add_all(
        [
            _presign("own", sales_user_id="u1", manager_user_id="other"),
            _presign("managed", sales_user_id="other", manager_user_id="u1"),
        ]
    )
    db.commit()

    ctx = MarketFieldContext("tenant-a", "u1", "specialist")
    rows = apply_presign_scope(
        db.query(BizPresignOrder), db, ctx, user_info=_user_info()
    ).all()

    assert _ids(rows) == {"own"}


def test_manager_matches_leader_region_or_document_manager():
    from app.services.market_field_scope import (
        MarketFieldContext,
        apply_presign_scope,
    )

    db = _session()
    db.add(
        BizCollectionRegionStaff(
            id="leader",
            region_id="region-led",
            user_id="u1",
            is_leader=1,
            tenant_no="tenant-a",
            create_dept_id="dept-a",
            del_flag=0,
        )
    )
    db.add_all(
        [
            _presign("by-region", region_id="region-led", manager_user_id="other"),
            _presign("by-manager", region_id="region-other", manager_user_id="u1"),
            _presign("outside", region_id="region-other", manager_user_id="other"),
        ]
    )
    db.commit()

    ctx = MarketFieldContext("tenant-a", "u1", "manager")
    rows = apply_presign_scope(
        db.query(BizPresignOrder), db, ctx, user_info=_user_info()
    ).all()

    assert _ids(rows) == {"by-region", "by-manager"}


def test_manager_ignores_non_leader_staff_rows():
    from app.services.market_field_scope import (
        MarketFieldContext,
        apply_presign_scope,
    )

    db = _session()
    db.add(
        BizCollectionRegionStaff(
            id="staff",
            region_id="region-staff",
            user_id="u1",
            is_leader=0,
            tenant_no="tenant-a",
            create_dept_id="dept-a",
            del_flag=0,
        )
    )
    db.add(_presign("staff-region", region_id="region-staff", manager_user_id="other"))
    db.commit()

    ctx = MarketFieldContext("tenant-a", "u1", "manager")

    assert (
        apply_presign_scope(
            db.query(BizPresignOrder), db, ctx, user_info=_user_info()
        ).all()
        == []
    )


def test_soft_deleted_or_cross_tenant_region_never_matches():
    from app.services.market_field_scope import (
        MarketFieldContext,
        apply_presign_scope,
    )

    db = _session()
    db.add_all(
        [
            BizCollectionRegionStaff(
                id="deleted-leader",
                region_id="region-deleted",
                user_id="u1",
                is_leader=1,
                tenant_no="tenant-a",
                create_dept_id="dept-a",
                del_flag=1,
            ),
            BizCollectionRegionStaff(
                id="other-tenant-leader",
                region_id="region-cross",
                user_id="u1",
                is_leader=1,
                tenant_no="tenant-b",
                create_dept_id="dept-a",
                del_flag=0,
            ),
            _presign("deleted-region", region_id="region-deleted", manager_user_id="other"),
            _presign("cross-region", region_id="region-cross", manager_user_id="other"),
            _presign("soft-deleted-doc", manager_user_id="u1", del_flag=1),
            _presign("cross-tenant-doc", tenant_no="tenant-b", manager_user_id="u1"),
        ]
    )
    db.commit()

    ctx = MarketFieldContext("tenant-a", "u1", "manager")

    assert (
        apply_presign_scope(
            db.query(BizPresignOrder), db, ctx, user_info=_user_info()
        ).all()
        == []
    )


def test_dual_role_contexts_do_not_union_specialist_and_manager_scope():
    from app.services.market_field_scope import (
        MarketFieldContext,
        apply_presign_scope,
    )

    db = _session()
    db.add_all(
        [
            _presign("specialist-only", sales_user_id="u1", manager_user_id="other"),
            _presign("manager-only", sales_user_id="other", manager_user_id="u1"),
        ]
    )
    db.commit()

    specialist = MarketFieldContext("tenant-a", "u1", "specialist")
    manager = MarketFieldContext("tenant-a", "u1", "manager")

    assert _ids(apply_presign_scope(
        db.query(BizPresignOrder), db, specialist, user_info=_user_info()
    ).all()) == {
        "specialist-only"
    }
    assert _ids(apply_presign_scope(
        db.query(BizPresignOrder), db, manager, user_info=_user_info()
    ).all()) == {
        "manager-only"
    }


def test_apply_scope_requires_general_data_scope_context():
    from app.services.market_field_scope import (
        MarketFieldContext,
        apply_presign_scope,
    )

    db = _session()
    ctx = MarketFieldContext("tenant-a", "u1", "specialist")

    with pytest.raises(TypeError):
        apply_presign_scope(db.query(BizPresignOrder), db, ctx)


@pytest.mark.parametrize(
    ("apply_name", "model"),
    [
        ("apply_presign_scope", BizPresignOrder),
        ("apply_cell_protocol_scope", BizCollectionProtocol),
        ("apply_donation_scope", BizResearchDonation),
    ],
)
@pytest.mark.parametrize(
    "user_info",
    [{}, {"user_id": ""}, {"user_id": "other", "role": "super_admin"}],
)
def test_apply_scopes_reject_missing_or_mismatched_user_identity(
    apply_name, model, user_info
):
    from app.services import market_field_scope

    db = _session()
    ctx = market_field_scope.MarketFieldContext("tenant-a", "u1", "specialist")
    apply_scope = getattr(market_field_scope, apply_name)

    with pytest.raises(PermissionError):
        apply_scope(db.query(model), db, ctx, user_info=user_info)


@pytest.mark.parametrize(
    "assert_name",
    [
        "assert_presign_access",
        "assert_cell_protocol_access",
        "assert_donation_access",
    ],
)
@pytest.mark.parametrize(
    "user_info",
    [{}, {"user_id": ""}, {"user_id": "other", "role": "super_admin"}],
)
def test_assert_access_rejects_missing_or_mismatched_user_identity(
    assert_name, user_info
):
    from app.services import market_field_scope

    db = _session()
    ctx = market_field_scope.MarketFieldContext("tenant-a", "u1", "specialist")
    assert_access = getattr(market_field_scope, assert_name)

    with pytest.raises(PermissionError):
        assert_access(db, "any-id", ctx, user_info=user_info)


def test_protocol_and_donation_queries_use_their_collection_info_scope():
    from app.services.market_field_scope import (
        MarketFieldContext,
        apply_cell_protocol_scope,
        apply_donation_scope,
    )

    db = _session()
    db.add_all(
        _cell_protocol("cell-own", sales_user_id="u1")
        + _cell_protocol("cell-other", sales_user_id="other")
        + _cell_protocol("cell-deleted-info", sales_user_id="u1", collection_del_flag=1)
        + _donation("donation-own", sales_user_id="u1")
        + _donation("donation-other", sales_user_id="other")
        + _donation("donation-deleted-info", sales_user_id="u1", collection_del_flag=1)
    )
    db.commit()

    ctx = MarketFieldContext("tenant-a", "u1", "specialist")

    assert _ids(
        apply_cell_protocol_scope(
            db.query(BizCollectionProtocol), db, ctx, user_info=_user_info()
        ).all()
    ) == {"cell-own"}
    assert _ids(
        apply_donation_scope(
            db.query(BizResearchDonation), db, ctx, user_info=_user_info()
        ).all()
    ) == {"donation-own"}


def test_cell_and_donation_manager_scope_uses_leader_region_or_document_manager():
    from app.services.market_field_scope import (
        MarketFieldContext,
        apply_cell_protocol_scope,
        apply_donation_scope,
    )

    db = _session()
    db.add_all(
        [
            BizCollectionRegionStaff(
                id="leader",
                region_id="region-led",
                user_id="u1",
                is_leader=1,
                tenant_no="tenant-a",
                create_dept_id="dept-a",
                del_flag=0,
            ),
            BizCollectionRegionStaff(
                id="non-leader",
                region_id="region-staff",
                user_id="u1",
                is_leader=0,
                tenant_no="tenant-a",
                create_dept_id="dept-a",
                del_flag=0,
            ),
        ]
        + _cell_protocol("cell-region", region_id="region-led", manager_user_id="other")
        + _cell_protocol("cell-manager", region_id="other", manager_user_id="u1")
        + _cell_protocol("cell-non-leader", region_id="region-staff", manager_user_id="other")
        + _cell_protocol(
            "cell-cross-info",
            collection_tenant_no="tenant-b",
            region_id="region-led",
            manager_user_id="u1",
        )
        + _cell_protocol("cell-deleted-parent", manager_user_id="u1", parent_del_flag=1)
        + _cell_protocol("cell-cross-parent", tenant_no="tenant-b", manager_user_id="u1")
        + _donation("donation-region", region_id="region-led", manager_user_id="other")
        + _donation("donation-manager", region_id="other", manager_user_id="u1")
        + _donation("donation-non-leader", region_id="region-staff", manager_user_id="other")
        + _donation(
            "donation-cross-info",
            collection_tenant_no="tenant-b",
            region_id="region-led",
            manager_user_id="u1",
        )
        + _donation("donation-deleted-parent", manager_user_id="u1", parent_del_flag=1)
        + _donation("donation-cross-parent", tenant_no="tenant-b", manager_user_id="u1")
    )
    db.commit()
    ctx = MarketFieldContext("tenant-a", "u1", "manager")

    assert _ids(
        apply_cell_protocol_scope(
            db.query(BizCollectionProtocol), db, ctx, user_info=_user_info()
        ).all()
    ) == {"cell-region", "cell-manager"}
    assert _ids(
        apply_donation_scope(
            db.query(BizResearchDonation), db, ctx, user_info=_user_info()
        ).all()
    ) == {"donation-region", "donation-manager"}


def test_assert_access_uses_the_same_scope_for_all_document_types():
    from app.services.market_field_scope import (
        MarketFieldContext,
        assert_cell_protocol_access,
        assert_donation_access,
        assert_presign_access,
    )

    db = _session()
    db.add_all(
        [
            _presign("presign-own", sales_user_id="u1"),
            _presign("presign-outside", sales_user_id="other"),
            _presign("presign-cross", tenant_no="tenant-b", sales_user_id="u1"),
            _presign("presign-deleted", sales_user_id="u1", del_flag=1),
        ]
    )
    db.add_all(_cell_protocol("cell-own", sales_user_id="u1"))
    db.add_all(_cell_protocol("cell-outside", sales_user_id="other"))
    db.add_all(_cell_protocol("cell-cross", tenant_no="tenant-b", sales_user_id="u1"))
    db.add_all(_cell_protocol("cell-deleted", sales_user_id="u1", parent_del_flag=1))
    db.add_all(_donation("donation-own", sales_user_id="u1"))
    db.add_all(_donation("donation-outside", sales_user_id="other"))
    db.add_all(_donation("donation-cross", tenant_no="tenant-b", sales_user_id="u1"))
    db.add_all(_donation("donation-deleted", sales_user_id="u1", parent_del_flag=1))
    db.commit()
    ctx = MarketFieldContext("tenant-a", "u1", "specialist")

    assert assert_presign_access(
        db, "presign-own", ctx, user_info=_user_info()
    ).id == "presign-own"
    assert assert_cell_protocol_access(
        db, "cell-own", ctx, user_info=_user_info()
    ).id == "cell-own"
    assert assert_donation_access(
        db, "donation-own", ctx, user_info=_user_info()
    ).id == "donation-own"

    rejected = {
        assert_presign_access: [
            "presign-outside",
            "presign-cross",
            "presign-deleted",
        ],
        assert_cell_protocol_access: [
            "cell-outside",
            "cell-cross",
            "cell-deleted",
        ],
        assert_donation_access: [
            "donation-outside",
            "donation-cross",
            "donation-deleted",
        ],
    }
    for checker, record_ids in rejected.items():
        for record_id in record_ids:
            with pytest.raises(HTTPException) as exc_info:
                checker(db, record_id, ctx, user_info=_user_info())
            assert exc_info.value.status_code == 404


def test_general_data_scope_and_field_scope_are_intersected():
    from app.services.market_field_scope import (
        MarketFieldContext,
        apply_presign_scope,
    )

    db = _session()
    db.add_all(
        [
            SysRole(
                id="self-role",
                role_name="仅本人",
                data_scope="4",
                status="1",
                tenant_no="tenant-a",
                create_dept_id="dept-a",
                del_flag=0,
            ),
            SysUserRole(
                id="user-role",
                user_id="u1",
                role_id="self-role",
                tenant_no="tenant-a",
                create_dept_id="dept-a",
                del_flag=0,
            ),
            _presign("both", sales_user_id="u1", create_by="u1"),
            _presign("field-only", sales_user_id="u1", create_by="other"),
            _presign("data-only", sales_user_id="other", create_by="u1"),
        ]
    )
    db.commit()
    ctx = MarketFieldContext("tenant-a", "u1", "specialist")

    rows = apply_presign_scope(
        db.query(BizPresignOrder),
        db,
        ctx,
        user_info={"user_id": "u1", "role": "admin", "dept_id": "dept-a"},
    ).all()

    assert _ids(rows) == {"both"}


def test_multi_role_union_is_still_intersected_with_field_scope():
    from app.services.market_field_scope import (
        MarketFieldContext,
        apply_presign_scope,
    )

    db = _session()
    db.add_all(
        [
            SysRole(
                id="self-role",
                role_name="仅本人",
                data_scope="4",
                status="1",
                tenant_no="tenant-a",
                create_dept_id="dept-a",
                del_flag=0,
            ),
            SysRole(
                id="all-role",
                role_name="全部数据",
                data_scope="1",
                status="1",
                tenant_no="tenant-a",
                create_dept_id="dept-a",
                del_flag=0,
            ),
            SysUserRole(
                id="self-user-role",
                user_id="u1",
                role_id="self-role",
                tenant_no="tenant-a",
                create_dept_id="dept-a",
                del_flag=0,
            ),
            SysUserRole(
                id="all-user-role",
                user_id="u1",
                role_id="all-role",
                tenant_no="tenant-a",
                create_dept_id="dept-a",
                del_flag=0,
            ),
            _presign("both", sales_user_id="u1", create_by="u1"),
            _presign("field-only", sales_user_id="u1", create_by="other"),
            _presign("data-only", sales_user_id="other", create_by="u1"),
        ]
    )
    db.commit()
    ctx = MarketFieldContext("tenant-a", "u1", "specialist")

    rows = apply_presign_scope(
        db.query(BizPresignOrder),
        db,
        ctx,
        user_info=_user_info(role="admin"),
    ).all()

    assert _ids(rows) == {"both", "field-only"}


def test_market_field_scope_index_migration_is_idempotent_and_complete():
    migration = (
        Path(__file__).resolve().parents[2]
        / "docs/sql/migrations/2026-10-10-market-field-scope-indexes.sql"
    )
    sql = migration.read_text(encoding="utf-8")
    compact = re.sub(r"\s+", " ", re.sub(r"--[^\n]*", "", sql)).lower()
    expected = {
        "idx_crs_market_leader_scope": (
            "biz_collection_region_staff",
            "tenant_no,user_id,is_leader,del_flag,region_id",
        ),
        "idx_po_market_sales_scope": (
            "biz_presign_order",
            "tenant_no,del_flag,sales_user_id",
        ),
        "idx_po_market_region_scope": (
            "biz_presign_order",
            "tenant_no,del_flag,region_id",
        ),
        "idx_po_market_manager_scope": (
            "biz_presign_order",
            "tenant_no,del_flag,region_manager_user_id",
        ),
        "idx_pci_market_parent_scope": (
            "biz_protocol_collection_info",
            "tenant_no,del_flag,collection_protocol_id",
        ),
        "idx_pci_market_sales_scope": (
            "biz_protocol_collection_info",
            "tenant_no,del_flag,sales_user_id,collection_protocol_id",
        ),
        "idx_pci_market_region_scope": (
            "biz_protocol_collection_info",
            "tenant_no,del_flag,region_id,collection_protocol_id",
        ),
        "idx_pci_market_manager_scope": (
            "biz_protocol_collection_info",
            "tenant_no,del_flag,region_manager_user_id,collection_protocol_id",
        ),
        "idx_rdc_market_parent_scope": (
            "biz_research_donation_collection",
            "tenant_no,del_flag,research_donation_id",
        ),
        "idx_rdc_market_sales_scope": (
            "biz_research_donation_collection",
            "tenant_no,del_flag,sales_user_id,research_donation_id",
        ),
        "idx_rdc_market_region_scope": (
            "biz_research_donation_collection",
            "tenant_no,del_flag,region_id,research_donation_id",
        ),
        "idx_rdc_market_manager_scope": (
            "biz_research_donation_collection",
            "tenant_no,del_flag,region_manager_user_id,research_donation_id",
        ),
    }

    assert compact.count("information_schema.statistics") == len(expected)
    assert compact.count("prepare stmt from @sql") == len(expected)
    assert compact.count("deallocate prepare stmt") == len(expected)
    for index_name, (table_name, columns) in expected.items():
        assert f"table_name = '{table_name}'" in compact
        assert f"index_name = '{index_name}'" in compact
        column_sql = ",".join(f"`{column}`" for column in columns.split(","))
        assert re.search(
            rf"alter table `{table_name}` add index `{index_name}` \({column_sql}\)",
            compact,
        )