"""真正经过 HTTP、文件、数据库、LangGraph 和 outbox 的业务闭环测试。"""
from datetime import timedelta
import json
from sqlalchemy import select
from app.domain.models import AgentConfig, AuditRule, AuditRun, ApprovalChain, ApprovalRecord, ApprovalRound, OutboxEvent, Reimbursement
from app.infrastructure.database import transaction, Session
from app.application.worker import worker
from app.application.approvals import start_approval
from app.shared.common import utcnow
from .helpers import BASE, fields, image_bytes, reimbursement, submit

def test_auth_rotation_and_access_control(client, headers):
    login = client.post(BASE + '/auth/login', json={'username': 'alice', 'password': 'Demo@2026!'}).json()['data']
    refresh = client.post(BASE + '/auth/refresh', json={'refresh_token': login['refresh_token']})
    assert refresh.status_code == 200
    assert client.post(BASE + '/auth/refresh', json={'refresh_token': login['refresh_token']}).status_code == 401
    assert client.get(BASE + '/agents', headers=headers['alice']).status_code == 403
    assert client.get(BASE + '/users', headers=headers['carol']).status_code == 403
    assert client.get(BASE + '/reimbursements').status_code == 401
    assert client.post(BASE + '/auth/login', json={'username': 'alice', 'password': 'wrong'}).status_code == 401

def test_upload_and_manual_approval_feedback_alert(client, headers, drain):
    row = reimbursement(client, headers)
    audit_id = submit(client, headers, row)
    # 同一幂等键和旧 revision 重送仍返回同一审核，不能新建一个重复任务。
    assert submit(client, headers, row) == audit_id
    drain()
    audit = client.get(f'{BASE}/audit/{audit_id}', headers=headers['alice']).json()['data']
    assert audit['status'] == 'completed' and audit['decision'] == 'to_supervisor'
    assert audit['result']['risk_degraded'] is True
    assert audit['result']['extracted_invoices'][0]['source'] == 'user_input_fallback'
    assert {node['node_name'] for node in audit['nodes']} == {'extract', 'rule_check', 'risk_detect', 'make_decision'}
    assert client.get(f'{BASE}/audit/{audit_id}/events', headers=headers['alice']).text.count('event: node') == 4
    records = client.get(BASE + '/approvals?pending=true', headers=headers['bob']).json()['data']['items']
    record = records[0]
    assert client.post(f"{BASE}/approvals/{record['id']}/decide", headers=headers['alice'], json={'action': 'approve', 'opinion': '本人签署'}).status_code == 403
    body = {'action': 'approve', 'opinion': '已检查发票原件与报销用途'}
    result = client.post(f"{BASE}/approvals/{record['id']}/decide", headers=headers['bob'], json=body)
    assert result.status_code == 200 and result.json()['data']['reimbursement']['status'] == 'approved'
    assert client.post(f"{BASE}/approvals/{record['id']}/decide", headers=headers['bob'], json=body).status_code == 409
    feedback = client.post(f'{BASE}/audit/{audit_id}/feedback', headers=headers['bob'], json={'approval_record_id': record['id'], 'corrected_fields': {'note': '人工核验'}, 'reason': '人工确认票面一致'})
    assert feedback.status_code == 200 and feedback.json()['data']['model_update_needed'] is True
    alerts = client.get(BASE + '/alerts', headers=headers['carol']).json()['data']['items']
    alert = alerts[0]
    assert client.post(f"{BASE}/alerts/{alert['id']}/resolve", headers=headers['carol'], json={'opinion': '已完成核验'}).status_code == 409
    assert client.post(f"{BASE}/alerts/{alert['id']}/acknowledge", headers=headers['carol'], json={'opinion': '接手核验'}).status_code == 200
    assert client.post(f"{BASE}/alerts/{alert['id']}/resolve", headers=headers['carol'], json={'opinion': '已核验原件，模型待配置'}).status_code == 200
    assert client.get(BASE + '/logs?trace=audit-' + str(audit_id), headers=headers['carol']).json()['data']['total'] >= 7
    assert client.get(BASE + '/stats/overview', headers=headers['alice']).json()['data']['approved_count'] == 1

def test_missing_fields_to_manual(client, headers, drain):
    row = reimbursement(client, headers, declared={})
    audit_id = submit(client, headers, row)
    drain()
    audit = client.get(f'{BASE}/audit/{audit_id}', headers=headers['alice']).json()['data']
    assert audit['decision'] == 'to_manual'
    assert client.get(BASE + '/approvals?pending=true', headers=headers['carol']).json()['data']['total'] == 1

def test_duplicate_is_critical_with_real_history(client, headers, drain):
    first = reimbursement(client, headers)
    submit(client, headers, first); drain()
    second = reimbursement(client, headers)
    audit_id = submit(client, headers, second); drain()
    audit = client.get(f'{BASE}/audit/{audit_id}', headers=headers['carol']).json()['data']
    duplicate = next(item for item in audit['rules'] if item['rule_code'] == 'invoice_no_duplicate')
    assert not duplicate['passed'] and duplicate['severity'] == 'critical'
    assert audit['decision'] == 'to_finance' and audit['risk_level'] == 3
    assert audit['snapshot']['context']['recent_submissions'] == 1
    assert audit['result']['risk_features']['dept_avg_amount_ratio'] == 1

def test_revision_foreign_owner_and_withdraw(client, headers, drain):
    row = reimbursement(client, headers)
    assert client.get(f"{BASE}/reimbursements/{row['id']}", headers=headers['erin']).status_code == 200
    assert client.patch(f"{BASE}/reimbursements/{row['id']}", headers=headers['erin'], json={'revision': row['revision'], 'title': '管理员覆盖'}).status_code == 403
    assert client.patch(f"{BASE}/reimbursements/{row['id']}", headers=headers['alice'], json={'revision': 1, 'title': '旧版本覆盖'}).status_code == 409
    audit_id = submit(client, headers, row)
    pending = 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': pending['revision']}).status_code == 200
    drain()
    assert client.get(f'{BASE}/audit/{audit_id}', headers=headers['alice']).json()['data']['status'] == 'cancelled'
    assert client.get(BASE + '/approvals', headers=headers['bob']).json()['data']['total'] == 0

def test_partial_upload_keeps_committed_file(client, headers):
    row = client.post(BASE + '/reimbursements', headers=headers['alice'], json={'title': '部分失败测试', 'total_amount': '106.00'}).json()['data']
    response = client.post(f"{BASE}/reimbursements/{row['id']}/invoices", headers=headers['alice'], data={'revision': row['revision']}, files=[('files', ('ok.png', image_bytes(), 'image/png')), ('files', ('fake.png', b'not-image', 'image/png'))])
    assert 'event: progress' in response.text and 'event: error' in response.text and 'event: completed' not in response.text
    detail = client.get(f"{BASE}/reimbursements/{row['id']}", headers=headers['alice']).json()['data']
    assert len(detail['invoices']) == 1 and detail['revision'] == row['revision'] + 1

def test_frozen_rules_return_and_resubmit(client, headers, drain):
    row = reimbursement(client, headers)
    first_id = submit(client, headers, row)
    rules = client.get(BASE + '/rules', headers=headers['erin']).json()['data']['items']
    rule = next(item for item in rules if item['code'] == 'tax_number_valid')
    update = client.patch(BASE + '/rules/tax_number_valid', headers=headers['erin'], json={'expected_version': rule['version'], 'params': {'lengths': [20]}})
    assert update.status_code == 200
    drain()
    first = client.get(f'{BASE}/audit/{first_id}', headers=headers['alice']).json()['data']
    assert next(item for item in first['rules'] if item['rule_code'] == 'tax_number_valid')['rule_version'] == 1
    approval = client.get(BASE + '/approvals?pending=true', headers=headers['bob']).json()['data']['items'][0]
    assert client.post(f"{BASE}/approvals/{approval['id']}/decide", headers=headers['bob'], json={'action': 'return', 'opinion': '补充业务说明'}).status_code == 200
    returned = client.get(f"{BASE}/reimbursements/{row['id']}", headers=headers['alice']).json()['data']
    second_id = submit(client, headers, returned, key='another-submit-key'); drain()
    second = client.get(f'{BASE}/audit/{second_id}', headers=headers['alice']).json()['data']
    assert second['run_no'] == 2 and next(item for item in second['rules'] if item['rule_code'] == 'tax_number_valid')['rule_version'] == 2
    assert client.post(f"{BASE}/approvals/{approval['id']}/decide", headers=headers['bob'], json={'action': 'approve', 'opinion': '旧节点重签'}).status_code == 409

def test_real_model_contract_mock_for_auto_approval(client, headers, drain, monkeypatch):
    from app.infrastructure.ai import ai
    from app.shared.config import settings
    from app.shared.schemas import ExtractedInvoice, RiskAssessment
    monkeypatch.setattr(settings, 'vlm_base_url', 'http://model-contract.test/v1')
    monkeypatch.setattr(settings, 'vlm_model', 'contract-test')
    async def contract(prompt, schema, **kwargs):
        if schema is ExtractedInvoice:
            return schema(**fields()), {'total_tokens': 10}
        return RiskAssessment(risk_level=1, risk_score='0.1', risk_factors=[], reasoning='测试模型返回低风险'), {'total_tokens': 8}
    monkeypatch.setattr(ai, 'structured', contract)
    rule = next(item for item in client.get(BASE + '/rules', headers=headers['erin']).json()['data']['items'] if item['code'] == 'policy_compliance')
    assert client.patch(BASE + '/rules/policy_compliance', headers=headers['erin'], json={'expected_version': rule['version'], 'params': {'required': False}}).status_code == 200
    row = reimbursement(client, headers)
    audit_id = submit(client, headers, row); drain()
    assert client.get(f'{BASE}/audit/{audit_id}', headers=headers['alice']).json()['data']['decision'] == 'auto_approve'
    detail = client.get(f"{BASE}/reimbursements/{row['id']}", headers=headers['alice']).json()['data']
    assert detail['status'] == 'approved' and detail['approvals'][0]['approver_id'] is None
    # 重复派单不产生第二审批轮次。
    client.portal.call(start_approval, audit_id)
    assert len(client.get(f"{BASE}/reimbursements/{row['id']}", headers=headers['alice']).json()['data']['approval_rounds']) == 1

def test_outbox_recovery_and_timeout_dedup(client, headers, drain):
    row = reimbursement(client, headers)
    audit_id = submit(client, headers, row)
    async def abandoned():
        async with transaction() as session:
            event = await session.scalar(select(OutboxEvent).where(OutboxEvent.event_type == 'audit'))
            event.status, event.lease_until = 'processing', utcnow() - timedelta(seconds=1)
    client.portal.call(abandoned); drain()
    async def overdue():
        async with transaction() as session:
            record = await session.scalar(select(ApprovalRecord))
            record.created_at = utcnow() - timedelta(hours=49)
    client.portal.call(overdue)
    client.portal.call(worker.scan_timeouts); client.portal.call(worker.scan_timeouts)
    assert client.get(BASE + '/alerts?type=approval_timeout', headers=headers['carol']).json()['data']['total'] == 1

def test_self_review_chain_missing_and_configuration_recovery(client, headers, drain):
    # carol 自己发起且关键字段缺失，默认人工链 carol 必须被拒绝。
    row = reimbursement(client, headers, declared={}, owner='carol')
    audit_id = submit(client, headers, row, owner='carol'); drain()
    assert client.get(BASE + '/alerts?type=chain_missing', headers=headers['carol']).json()['data']['total'] == 1
    assert client.get(BASE + '/approvals?pending=true', headers=headers['carol']).json()['data']['total'] == 0
    # 派单耗尽后配置新的非提交审核员，创建链会重新激活持久事件。
    async def exhausted():
        async with transaction() as session:
            event = await session.scalar(select(OutboxEvent).where(OutboxEvent.event_type == 'approval'))
            event.status = 'failed'
    client.portal.call(exhausted)
    user = client.post(BASE + '/users', headers=headers['erin'], json={'username': 'frank', 'real_name': '另一审计员', 'password': 'Demo@2026!', 'department_id': 1, 'role': 'auditor'}).json()['data']
    assert client.post(BASE + '/approvals/chains', headers=headers['dave'], json={'name': '替代人工复核链', 'route': 'to_manual', 'priority': 100, 'nodes': [{'approver_id': user['id'], 'role': 'auditor'}]}).status_code == 201
    drain()
    detail = client.get(f"{BASE}/reimbursements/{row['id']}", headers=headers['carol']).json()['data']
    assert detail['status'] == 'in_approval' and detail['approvals'][0]['approver_id'] == user['id']

def test_config_optimistic_and_core_guards(client, headers):
    assert client.patch(BASE + '/agents/rule_check', headers=headers['erin'], json={'expected_version': 1, 'enabled': False}).status_code == 422
    assert client.patch(BASE + '/rules/invoice_no_duplicate', headers=headers['erin'], json={'expected_version': 1, 'severity': 'warning'}).status_code == 422
    body = {'expected_version': 1, 'config': {'max_attempts': 2, 'timeout': 10, 'concurrency': 2}}
    assert client.patch(BASE + '/agents/extract', headers=headers['erin'], json=body).status_code == 200
    assert client.patch(BASE + '/agents/extract', headers=headers['erin'], json=body).status_code == 409
    assert client.post(BASE + '/agents/extract/test', headers=headers['carol']).json()['data']['status'] == 'degraded'
    assert client.patch(BASE + '/users/5', headers=headers['erin'], json={'active': False}).status_code == 409

def test_policies_keyword_versions_and_exports(client, headers):
    row = client.post(BASE + '/knowledge/documents', headers=headers['carol'], json={'title': '测试制度原文', 'content': '# 第一条\n\n测试制度：差旅费用应说明用途。'}).json()['data']
    result = client.post(BASE + '/knowledge/search', headers=headers['alice'], json={'query': '差旅', 'mode': 'keyword'})
    assert result.status_code == 200 and result.json()['data']['items'][0]['document_id'] == row['id']
    assert client.post(BASE + '/knowledge/search', headers=headers['alice'], json={'query': '差旅', 'mode': 'semantic'}).status_code == 503
    assert client.post(f"{BASE}/knowledge/documents/{row['id']}/reindex", headers=headers['carol']).status_code == 503
    assert client.patch(f"{BASE}/knowledge/documents/{row['id']}", headers=headers['carol'], json={'title': '测试制度修订', 'content': '测试办公政策', 'expected_version': 1}).json()['data']['version'] == 2
    reimbursement(client, headers)
    assert client.get(BASE + '/stats/export?format=xlsx', headers=headers['dave']).content[:2] == b'PK'
    assert client.get(BASE + '/stats/export?format=pdf', headers=headers['dave']).content[:4] == b'%PDF'
    assert client.get(BASE + '/logs/export', headers=headers['carol']).content.startswith(b'\xef\xbb\xbf')
    assert client.get(BASE + '/stats/export?format=xlsx', headers=headers['alice']).status_code == 403

def test_multi_node_freeze_and_identity_change(client, headers, drain):
    # 同具体度、高优先级的两节点主管链覆盖默认链。
    chain = client.post(BASE + '/approvals/chains', headers=headers['dave'], json={'name': '两级测试链', 'route': 'to_supervisor', 'priority': 90,
        'nodes': [{'approver_id': 2, 'role': 'supervisor'}, {'approver_id': 4, 'role': 'manager'}]}).json()['data']
    row = reimbursement(client, headers); submit(client, headers, row); drain()
    # 修改当前链不能改变正在进行的第二审批人。
    assert client.patch(f"{BASE}/approvals/chains/{chain['id']}", headers=headers['dave'], json={'name': '改为单节点', 'route': 'to_supervisor', 'priority': 90, 'nodes': [{'approver_id': 2, 'role': 'supervisor'}], 'expected_version': 1}).status_code == 200
    bob = client.get(BASE + '/approvals?pending=true', headers=headers['bob']).json()['data']['items'][0]
    assert client.post(f"{BASE}/approvals/{bob['id']}/decide", headers=headers['bob'], json={'action': 'approve', 'opinion': '第一节点通过'}).status_code == 200
    dave = client.get(BASE + '/approvals?pending=true', headers=headers['dave']).json()['data']['items'][0]
    assert client.patch(BASE + '/users/4', headers=headers['erin'], json={'role': 'auditor'}).status_code == 200
    assert client.post(f"{BASE}/approvals/{dave['id']}/decide", headers=headers['dave'], json={'action': 'approve', 'opinion': '身份变化后签署'}).status_code == 409
    assert client.patch(BASE + '/users/4', headers=headers['erin'], json={'role': 'manager'}).status_code == 200
    assert client.post(f"{BASE}/approvals/{dave['id']}/decide", headers=headers['dave'], json={'action': 'reject', 'opinion': '第二节点驳回'}).status_code == 200
    assert client.get(f"{BASE}/reimbursements/{row['id']}", headers=headers['alice']).json()['data']['status'] == 'rejected'