"""验证并发单号、文件页数、异步取消、外部重试与授权边界。"""
from concurrent.futures import ThreadPoolExecutor
from datetime import date
from sqlalchemy import select
import httpx
import pymupdf
import pytest
from app.domain.models import OutboxEvent
from app.infrastructure.database import transaction
from app.infrastructure.files import validate_file
from app.infrastructure.ai import ai, AIUnavailable
from app.shared.common import BusinessError
from app.shared.config import settings
from .helpers import BASE, fields, reimbursement, submit

def test_parallel_number_allocation(client, headers):
    def create(index):
        response = client.post(BASE + '/reimbursements', headers=headers['alice'], json={'title': f'并发测试{index}', 'total_amount': '1.00'})
        assert response.status_code == 201, response.text
        return response.json()['data']['number']
    with ThreadPoolExecutor(max_workers=4) as pool:
        numbers = list(pool.map(create, range(8)))
    assert len(set(numbers)) == 8

def test_pdf_type_and_page_limit():
    with pymupdf.open() as pdf:
        pdf.new_page()
        assert validate_file(pdf.tobytes()) == ('application/pdf', '.pdf')
        for _ in range(5):
            pdf.new_page()
        with pytest.raises(BusinessError) as error:
            validate_file(pdf.tobytes())
        assert error.value.code == 'PDF_LIMIT'

def test_graph_extraction_retry_is_bounded(client, headers, drain, monkeypatch):
    monkeypatch.setattr(settings, 'vlm_base_url', 'http://transient.test/v1')
    monkeypatch.setattr(settings, 'vlm_model', 'transient-test')
    attempts = []
    async def failing(prompt, schema, **kwargs):
        if kwargs.get('vision'):
            attempts.append(1)
        raise AIUnavailable('瞬时网络故障')
    monkeypatch.setattr(ai, 'structured', failing)
    row = reimbursement(client, headers)
    audit_id = submit(client, headers, row); drain()
    audit = client.get(f'{BASE}/audit/{audit_id}', headers=headers['carol']).json()['data']
    assert len(attempts) == 3
    assert len([node for node in audit['nodes'] if node['node_name'] == 'extract']) == 3
    assert audit['status'] == 'completed' and audit['result']['extracted_invoices'][0]['source'] == 'user_input_fallback'

def test_cancelled_audit_not_overwritten_by_late_failure(client, headers, drain, monkeypatch):
    from app.application import pipeline
    row = reimbursement(client, headers)
    audit_id = submit(client, headers, row)
    current = client.get(f"{BASE}/reimbursements/{row['id']}", headers=headers['alice']).json()['data']
    assert client.post(f"{BASE}/reimbursements/{row['id']}/withdraw", headers=headers['alice'], json={'revision': current['revision']}).status_code == 200
    async def exhaust():
        async with transaction() as session:
            event = await session.scalar(select(OutboxEvent).where(OutboxEvent.event_type == 'audit'))
            event.attempts = 2
    async def fail_late(audit):
        raise RuntimeError('明确模拟撤回后的迟到核心失败')
    client.portal.call(exhaust)
    monkeypatch.setattr(pipeline, 'run_audit', fail_late)
    drain()
    assert client.get(f'{BASE}/audit/{audit_id}', headers=headers['alice']).json()['data']['status'] == 'cancelled'

def test_gross_total_mismatch_requires_finance(client, headers, drain):
    row = reimbursement(client, headers, total='100.00')
    audit_id = submit(client, headers, row); drain()
    audit = client.get(f'{BASE}/audit/{audit_id}', headers=headers['alice']).json()['data']
    assert audit['decision'] == 'to_finance'
    assert not next(item for item in audit['rules'] if item['rule_code'] == 'reimbursement_total_match')['passed']

def test_cross_department_and_cors(client, headers):
    department = client.post(BASE + '/departments', headers=headers['erin'], json={'name': '另一测试部门'}).json()['data']
    user = client.post(BASE + '/users', headers=headers['erin'], json={'username': 'outside', 'real_name': '跨部门员工', 'password': 'Demo@2026!', 'department_id': department['id'], 'role': 'employee'}).json()['data']
    token = client.post(BASE + '/auth/login', json={'username': 'outside', 'password': 'Demo@2026!'}).json()['data']['access_token']
    row = client.post(BASE + '/reimbursements', headers={'Authorization': 'Bearer ' + token}, json={'title': '跨部门测试', 'total_amount': '1.00'}).json()['data']
    assert client.get(f"{BASE}/reimbursements/{row['id']}", headers=headers['alice']).status_code == 403
    assert client.get(f"{BASE}/reimbursements/{row['id']}", headers=headers['bob']).status_code == 403
    assert client.get(f"{BASE}/reimbursements/{row['id']}", headers=headers['carol']).status_code == 200
    response = client.options(BASE + '/reimbursements', headers={'Origin': 'http://127.0.0.1:5173', 'Access-Control-Request-Method': 'POST', 'Access-Control-Request-Headers': 'authorization,content-type,idempotency-key'})
    assert response.status_code == 200 and response.headers['access-control-allow-origin'] == 'http://127.0.0.1:5173'
    assert client.options(BASE + '/reimbursements', headers={'Origin': 'http://untrusted.test', 'Access-Control-Request-Method': 'POST'}).status_code == 400
    assert client.get('/openapi.json').json()['components']['securitySchemes']['HTTPBearer']['scheme'] == 'bearer'

def test_invalid_rule_test_context_is_422(client, headers):
    response = client.post(BASE + '/rules/invoice_date_valid/test', headers=headers['carol'], json={'invoice': {'invoice_date': '2026-01-01'}, 'context': {'today': 'not-a-date'}})
    assert response.status_code == 422 and response.json()['error']['code'] == 'RULE_TEST_INVALID'