fix: stabilize backend services and frontend pointer
This commit is contained in:
@@ -0,0 +1,68 @@
|
||||
import asyncio
|
||||
from unittest.mock import patch
|
||||
|
||||
from fastapi_modules.fastapi_leaudit.services.impl.contractTemplateServiceImpl import ContractTemplateServiceImpl
|
||||
|
||||
|
||||
class _EmptyMappingResult:
|
||||
def mappings(self):
|
||||
return self
|
||||
|
||||
def all(self):
|
||||
return []
|
||||
|
||||
|
||||
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"] == ["省级", "公共", "梅州"]
|
||||
@@ -0,0 +1,58 @@
|
||||
import pytest
|
||||
|
||||
from fastapi_modules.fastapi_leaudit.services.impl.pageQualityServiceImpl import PageQualityServiceImpl
|
||||
|
||||
|
||||
class _FakeVlmClient:
|
||||
def __init__(self, response):
|
||||
self.response = response
|
||||
self.prompts = []
|
||||
|
||||
async def extract_multifield(self, *, prompt, images_data_urls, max_tokens=800):
|
||||
self.prompts.append((prompt, images_data_urls, max_tokens))
|
||||
if isinstance(self.response, Exception):
|
||||
raise self.response
|
||||
return self.response
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_vlm_page_quality_reject_result_is_used():
|
||||
service = PageQualityServiceImpl()
|
||||
service.VlmClient = _FakeVlmClient(
|
||||
{
|
||||
"status": "reject",
|
||||
"score": 0.18,
|
||||
"reason": "页面文字严重模糊,无法稳定辨认关键内容",
|
||||
}
|
||||
)
|
||||
|
||||
status, score, reason = await service._classify_page_image_by_vlm(b"image-bytes")
|
||||
|
||||
assert status == "reject"
|
||||
assert score == 0.18
|
||||
assert "严重模糊" in reason
|
||||
assert "只输出 JSON" in service.VlmClient.prompts[0][0]
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_vlm_page_quality_invalid_result_falls_back_to_review_not_pass():
|
||||
service = PageQualityServiceImpl()
|
||||
service.VlmClient = _FakeVlmClient({"status": "unknown", "reason": ""})
|
||||
|
||||
status, score, reason = await service._classify_page_image_by_vlm(b"image-bytes")
|
||||
|
||||
assert status == "review"
|
||||
assert score == 0.5
|
||||
assert "VLM返回结果不可用" in reason
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_vlm_page_quality_error_falls_back_to_review_not_pass():
|
||||
service = PageQualityServiceImpl()
|
||||
service.VlmClient = _FakeVlmClient(RuntimeError("vlm down"))
|
||||
|
||||
status, score, reason = await service._classify_page_image_by_vlm(b"image-bytes")
|
||||
|
||||
assert status == "review"
|
||||
assert score == 0.5
|
||||
assert "VLM图片质量检测失败" in reason
|
||||
Reference in New Issue
Block a user