leaudit-platform-backend/tests/test_contract_template_search.py

import asyncio
from unittest.mock import patch

import pytest

from fastapi_common.fastapi_common_web.exception.LeauditException import LeauditException
from fastapi_modules.fastapi_leaudit.services.impl.contractTemplateServiceImpl import ContractTemplateServiceImpl


class _EmptyMappingResult:
    def mappings(self):
        return self

    def all(self):
        return []

    def first(self):
        return None


class _FakeSession:
    def __init__(self):
        self.executed_sql = None
        self.executed_params = None

    async def execute(self, sql, params=None):
        self.executed_sql = sql
        self.executed_params = params
        return _EmptyMappingResult()


class _FakeSessionContext:
    def __init__(self, session):
        self.session = session

    async def __aenter__(self):
        return self.session

    async def __aexit__(self, exc_type, exc, tb):
        return None


def test_contract_template_search_category_stats_binds_expanding_scope_params():
    service = ContractTemplateServiceImpl()
    fake_session = _FakeSession()

    async def noop_ensure_schema(session):
        return None

    async def fake_user_context(current_user_id, session):
        return {
            "id": current_user_id,
            "area": "梅州",
            "tenant_code": "MZ",
            "tenant_name": "梅州",
            "tenant_scope_value": "梅州",
            "is_global": False,
            "can_manage": True,
            "is_area_admin": True,
        }

    service._ensureContractTemplateSchema = noop_ensure_schema
    service._getCurrentUserContext = fake_user_context

    with patch(
        "fastapi_modules.fastapi_leaudit.services.impl.contractTemplateServiceImpl.GetAsyncSession",
        return_value=_FakeSessionContext(fake_session),
    ):
        asyncio.run(service._load_search_category_stats("买卖", None, None, 5))

    assert fake_session.executed_sql._bindparams["visible_tenant_codes"].expanding is True
    assert fake_session.executed_sql._bindparams["visible_regions"].expanding is True
    assert fake_session.executed_params["visible_tenant_codes"] == ["PROVINCIAL", "PUBLIC", "MZ"]
    assert fake_session.executed_params["visible_regions"] == ["省级", "公共", "梅州"]


def test_contract_template_upload_scope_allows_global_manager_with_tenant():
    service = ContractTemplateServiceImpl()

    tenant_code, tenant_name, region = service._resolve_upload_scope(
        {
            "id": 1,
            "area": "梅州",
            "tenant_code": "MZ",
            "tenant_name": "梅州",
            "tenant_scope_value": "梅州",
            "is_global": True,
            "can_manage": True,
            "is_area_admin": False,
        },
        "梅州",
        "MZ",
    )

    assert tenant_code == "MZ"
    assert tenant_name == "梅州"
    assert region == "梅州"


def test_contract_template_upload_scope_rejects_non_manager():
    service = ContractTemplateServiceImpl()

    with pytest.raises(LeauditException):
        service._resolve_upload_scope(
            {
                "id": 2,
                "area": "梅州",
                "tenant_code": "MZ",
                "tenant_name": "梅州",
                "tenant_scope_value": "梅州",
                "is_global": False,
                "can_manage": False,
                "is_area_admin": False,
            },
            "梅州",
            "MZ",
        )


def test_contract_template_detail_binds_expanding_scope_params():
    service = ContractTemplateServiceImpl()
    fake_session = _FakeSession()

    async def noop_ensure_schema(session):
        return None

    async def fake_user_context(current_user_id, session):
        return {
            "id": current_user_id,
            "area": "梅州",
            "tenant_code": "MZ",
            "tenant_name": "梅州",
            "tenant_scope_value": "梅州",
            "is_global": False,
            "can_manage": True,
            "is_area_admin": True,
        }

    service._ensureContractTemplateSchema = noop_ensure_schema
    service._getCurrentUserContext = fake_user_context

    with patch(
        "fastapi_modules.fastapi_leaudit.services.impl.contractTemplateServiceImpl.GetAsyncSession",
        return_value=_FakeSessionContext(fake_session),
    ):
        result = asyncio.run(service.GetTemplateDetail(32, 5))

    assert result is None
    assert fake_session.executed_sql._bindparams["visible_tenant_codes"].expanding is True
    assert fake_session.executed_sql._bindparams["visible_regions"].expanding is True
    assert fake_session.executed_params["visible_tenant_codes"] == ["PROVINCIAL", "PUBLIC", "MZ"]
    assert fake_session.executed_params["visible_regions"] == ["省级", "公共", "梅州"]


def test_contract_template_detail_permission_does_not_fallback_to_list_permission():
    from fastapi_modules.fastapi_leaudit.controllers.contractTemplateController import ContractTemplateController

    controller = ContractTemplateController()

    class FakePermissionService:
        async def CheckPermission(self, user_id, permission_key):
            return permission_key == "contract_template:list:read"

    controller.PermissionService = FakePermissionService()

    assert asyncio.run(controller._check_permission(5, ["contract_template:detail:read"])) is False