shanxia2.0/backend/tests/test_pl_quota_recognition.py

"""图片解析边界测试，不访问外部服务或数据库。"""
import base64
import importlib.util
import io
import json
from decimal import Decimal
from types import SimpleNamespace

import pytest
from fastapi import HTTPException
from PIL import Image


def test_recognition_module_exists():
    assert importlib.util.find_spec('app.services.pl_quota_recognition') is not None


def png():
    out = io.BytesIO()
    Image.new('RGB', (64, 64), 'white').save(out, 'PNG')
    return out.getvalue()


def configured(monkeypatch, payload, status=200):
    from app.services import pl_quota_recognition as svc
    monkeypatch.setattr(svc, 'settings', SimpleNamespace(
        DEEPSEEK_API_KEY='test-only', DEEPSEEK_BASE_URL='https://api.deepseek.com',
        DEEPSEEK_VISION_MODEL='deepseek-flash',
    ))
    calls = []
    def post(url, **kwargs):
        calls.append((url, kwargs))
        return SimpleNamespace(status_code=status, json=lambda: {
            'choices': [{'finish_reason': 'stop', 'message': {'content': json.dumps(payload)}}]
        })
    monkeypatch.setattr(svc.requests, 'post', post)
    return svc, calls


def test_candidate_fields_only_no_activation_or_link_ids(monkeypatch):
    svc, calls = configured(monkeypatch, {
        'certNo': 'G-1', 'totalQty': '12.123456', 'quantityUnit': '吨',
        'status': 'active', 'usedQty': 0, 'tenantNo': 'other', 'contractId': 'foreign',
    })
    result = svc.recognize_quota_cert('purchase', png())
    assert result['candidates'] == {'certNo': 'G-1', 'totalQty': Decimal('12.123456'), 'quantityUnit': '吨'}
    assert result['requiresConfirmation'] is True
    assert len(calls) == 1
    assert calls[0][1]['allow_redirects'] is False
    request = calls[0][1]['json']
    assert request['response_format'] == {'type': 'json_object'}
    assert request['messages'][1]['content'][1]['image_url']['url'].startswith('data:image/png;base64,')


def test_invalid_field_dropped_without_guessing_other_values(monkeypatch):
    svc, _ = configured(monkeypatch, {'totalQty': 'NaN', 'validTo': 'not-a-date', 'buyerName': '购方'})
    result = svc.recognize_quota_cert('purchase', png())
    assert result['candidates'] == {'buyerName': '购方'}
    assert set(result['invalidFields']) == {'totalQty', 'validTo'}


def test_transport_references_have_no_database_identity(monkeypatch):
    svc, _ = configured(monkeypatch, {'vehicleNumbersJson': ['皖A001', '皖A002'], 'purchaseRefs': [
        {'referenceType': 'purchase_permit', 'referenceNo': 'G-1', 'purchaseCertId': 'injected', 'id': 'injected'}
    ]})
    result = svc.recognize_quota_cert('transport', png())
    assert result['candidates']['purchaseRefs'] == [{'referenceType': 'purchase_permit', 'referenceNo': 'G-1'}]
    assert len(result['candidates']['vehicleNumbersJson']) == 2


def test_invalid_image_rejected_before_network(monkeypatch):
    svc, calls = configured(monkeypatch, {})
    with pytest.raises(HTTPException):
        svc.recognize_quota_cert('purchase', b'not an image')
    assert calls == []


def _pdf(page_count):
    """构造仅含 Helvetica 文本的未加密 PDF，供渲染测试使用。"""
    objects = {
        1: b'<< /Type /Catalog /Pages 2 0 R >>',
        2: ('<< /Type /Pages /Count %d /Kids [%s] >>' % (
            page_count, ' '.join('%d 0 R' % (3 + i * 2) for i in range(page_count))
        )).encode(),
    }
    for index in range(page_count):
        stream = ('BT /F1 18 Tf 36 120 Td (PAGE-%d) Tj ET' % (index + 1)).encode()
        objects[3 + index * 2] = (
            '<< /Type /Page /Parent 2 0 R /MediaBox [0 0 320 200] '
            '/Resources << /Font << /F1 << /Type /Font /Subtype /Type1 /BaseFont /Helvetica >> >> >> '
            '/Contents %d 0 R >>' % (4 + index * 2)
        ).encode()
        objects[4 + index * 2] = b'<< /Length %d >>\nstream\n%s\nendstream' % (len(stream), stream)
    out = bytearray(b'%PDF-1.4\n')
    offsets = [0]
    for number in range(1, len(objects) + 1):
        offsets.append(len(out))
        out += b'%d 0 obj\n%s\nendobj\n' % (number, objects[number])
    xref = len(out)
    out += b'xref\n0 %d\n' % len(offsets)
    out += b'0000000000 65535 f \n'
    for offset in offsets[1:]:
        out += b'%010d 00000 n \n' % offset
    out += b'trailer\n<< /Size %d /Root 1 0 R >>\nstartxref\n%d\n%%%%EOF\n' % (len(offsets), xref)
    return bytes(out)


def _image_parts(call):
    return [part for part in call[1]['json']['messages'][1]['content'] if part.get('type') == 'image_url']


def test_single_page_pdf_is_sent_as_png(monkeypatch):
    svc, calls = configured(monkeypatch, {'certNo': 'Y34260268310443'})
    result = svc.recognize_quota_cert('transport', _pdf(1))
    assert result['candidates']['certNo'] == 'Y34260268310443'
    images = _image_parts(calls[0])
    assert len(images) == 1
    encoded = images[0]['image_url']['url']
    assert encoded.startswith('data:image/png;base64,')
    with Image.open(io.BytesIO(base64.b64decode(encoded.split(',', 1)[1]))) as image:
        assert image.format == 'PNG'
        assert min(image.size) > 0


def test_multipage_pdf_keeps_page_order(monkeypatch):
    svc, calls = configured(monkeypatch, {'certNo': 'Y1'})
    svc.recognize_quota_cert('transport', _pdf(2))
    assert len(_image_parts(calls[0])) == 2
    assert '按页序' in calls[0][1]['json']['messages'][1]['content'][0]['text']


def test_pdf_over_five_pages_rejected_before_network(monkeypatch):
    svc, calls = configured(monkeypatch, {})
    with pytest.raises(HTTPException) as error:
        svc.recognize_quota_cert('transport', _pdf(6))
    assert calls == []
    assert '超过5页' in error.value.detail


def test_encrypted_pdf_rejected_before_network(monkeypatch):
    import fitz
    document = fitz.open()
    document.new_page()
    raw = document.tobytes(encryption=fitz.PDF_ENCRYPT_AES_256, user_pw='secret', owner_pw='secret')
    document.close()
    svc, calls = configured(monkeypatch, {})
    with pytest.raises(HTTPException) as error:
        svc.recognize_quota_cert('transport', raw)
    assert calls == []
    assert '加密' in error.value.detail


def test_partial_reference_recognition_keeps_readable_numbers(monkeypatch):
    svc, _ = configured(monkeypatch, {'purchaseRefs': [
        {'referenceType': 'purchase_permit', 'referenceNo': 'G-1'},
        {'referenceType': 'unknown', 'referenceNo': None},
        {'referenceType': None, 'referenceNo': 'Readable-2'},
    ]})
    result = svc.recognize_quota_cert('transport', png())
    assert result['candidates']['purchaseRefs'] == [
        {'referenceType': 'purchase_permit', 'referenceNo': 'G-1'},
        {'referenceType': 'unknown', 'referenceNo': 'Readable-2'},
    ]
    assert 'purchaseRefs[1]' in result['invalidFields']


@pytest.mark.parametrize('status', [401, 429, 500])
def test_provider_error_is_sanitized_and_allows_manual_entry(monkeypatch, status):
    svc, _ = configured(monkeypatch, {}, status)
    with pytest.raises(HTTPException) as error:
        svc.recognize_quota_cert('purchase', png())
    assert '手工' in error.value.detail
    assert 'test-only' not in error.value.detail


def test_merge_candidates_preserves_user_edits_and_zero():
    from app.services.pl_quota_recognition import merge_unedited_candidates
    assert merge_unedited_candidates({'totalQty': 0, 'buyerName': ''},
                                    {'totalQty': 100, 'buyerName': '识别名', 'sellerName': '销售方'},
                                    {'buyerName'}) == {
        'totalQty': 0, 'buyerName': '', 'sellerName': '销售方'
    }