"""测试工作台确实执行业务、隔离上传资料，并预览实际可下载的结果文件。"""
import asyncio
from copy import copy
from copy import deepcopy
from dataclasses import replace
import json
from pathlib import Path
from xml.etree import ElementTree as ET
from zipfile import ZipFile

import httpx
import pytest

from api.test_lab_routes import TestRequest as LabRequest
from db import business_store, db_conn, test_lab_store
from extensions.main_fan import service as main_fan_service
from main import app
from runtime import business_tasks
from runtime.business_tasks import BusinessTaskController
from runtime.test_lab import service
from runtime.test_lab.presets import main_fan_input
from runtime.test_lab.spreadsheets import NS, preview_workbook, write_workbook
from runtime.test_lab.main_fan_simulation import MainFanSimulation, build_input


@pytest.fixture
def controller(monkeypatch,tmp_path):
    # 下列旧规则/导出兼容用例显式挂载历史服务；生产默认没有 pipeline test_runner。
    from extensions import registry as extensions
    catalog = copy(extensions.load_extensions())
    catalog.business_agents = dict(catalog.business_agents)
    spec = catalog.business_agents['main_fan']
    catalog.business_agents['main_fan'] = replace(spec, test_runner=main_fan_service.analyze)
    monkeypatch.setattr(extensions, '_loaded', catalog)
    monkeypatch.setenv('TEST_LAB_ENABLED','1')
    monkeypatch.setenv('TEST_LAB_DATA_DIR',str(tmp_path/'assets'))
    monkeypatch.setenv('TEST_LAB_SAMPLE_DIR',str(tmp_path/'samples'))
    value=BusinessTaskController()
    monkeypatch.setattr(business_tasks,'_controller',value)
    return value


def request(scenario='normal',**extra):
    extra.setdefault('execution_mode', 'pipeline')  # 显式测试兼容 runner；默认 API 已是完整 Agent。
    return LabRequest(agent_name='main_fan',scenario_id=scenario,label=scenario,input=main_fan_input(scenario),**extra)


@pytest.mark.asyncio
@pytest.mark.parametrize('scenario,expected',[('normal','NORMAL'),('warning','WARNING'),('abnormal','ABNORMAL'),('insufficient','UNDETERMINED')])
async def test_runs_use_real_business_rules_and_preserve_request_and_result(controller,scenario,expected):
    value=service.start_run(request(scenario),'lab-user')
    job=business_store.claim('main_fan')
    assert job['id']==value['id']
    await controller.run_job(job)
    result=service.public_run(value['id'],'lab-user')
    assert result['status']=='completed',result['error']
    assert result['result']['platform_payload']['judgment']['resultCode']==expected
    assert result['result']['assessment']['decision']['result_code']==expected
    assert result['request']['input']['reports']==value['request']['input']['reports']
    assert result['request']['publish'] is False
    assert business_store.get(job['id'])['push_status']=='none'
    assert len(result['stages'])==7
    assert service.public_run(value['id'],'lab-user')==result


def test_test_request_retries_are_idempotent_and_changed_inputs_are_rejected(controller):
    payload=request(request_id='same-request')
    first=service.start_run(payload,'lab-user')
    second=service.start_run(payload,'lab-user')
    assert first['id']==second['id'] and first['session_id']==second['session_id']
    assert db_conn.scalar('SELECT COUNT(*) FROM test_lab_runs',())==1
    changed=payload.model_copy(deep=True)
    changed.input['reports'][1]['model']='DIFFERENT'
    with pytest.raises(ValueError,match='不同'):
        service.start_run(changed,'lab-user')


@pytest.mark.asyncio
async def test_upload_content_json_preview_and_cross_user_access(controller):
    transport=httpx.ASGITransport(app=app)
    async with httpx.AsyncClient(transport=transport,base_url='http://test') as http:
        response=await http.post('/api/test-lab/assets?user_name=alice',files={'file':('../资料.json',json.dumps(main_fan_input('normal'),ensure_ascii=False).encode(),'application/json')})
        assert response.status_code==201,response.text
        asset=response.json()
        assert asset['filename']=='资料.json'
        assert (await http.get(asset['preview_url']+'?user_name=alice')).json()['content']['fan_id']=='mock-report-fan-1'
        downloaded=await http.get(asset['url']+'?user_name=alice')
        assert downloaded.status_code==200
        assert downloaded.headers['x-content-type-options']=='nosniff'
        assert (await http.get(asset['url']+'?user_name=bob')).status_code==404
        assert (await http.get('/api/test-lab/assets/not-an-id/content?user_name=alice')).status_code==404
        assert (await http.post('/api/test-lab/assets',files={'file':('bad.json',b'{bad','application/json')})).status_code==400
        assert (await http.post('/api/test-lab/assets',files={'file':('code.html',b'<script>alert(1)</script>','text/html')})).status_code==400


@pytest.mark.asyncio
async def test_pdf_selection_does_not_mix_synthetic_report_facts(controller,tmp_path):
    transport=httpx.ASGITransport(app=app)
    async with httpx.AsyncClient(transport=transport,base_url='http://test') as http:
        upload=await http.post('/api/test-lab/assets',files={'file':('测试.pdf',b'%PDF-1.4\nsynthetic unit-test file','application/pdf')})
        identifier=upload.json()['id']
        payload=request(asset_ids=[identifier]).model_dump(mode='json')
        response=await http.post('/api/test-lab/runs',json=payload)
        assert response.status_code==400 and '混用' in response.text
        payload=request('documents',asset_ids=[identifier]).model_dump(mode='json')
        response=await http.post('/api/test-lab/runs',json=payload)
        assert response.status_code==202,response.text
        created=response.json()
        assert created['request']['input']['reports']==[]
        assert len(created['request']['input']['materials'])==1
        assert created['request']['assets'][0]['sha256']
        assert 'tmp_path' not in response.text
        record=business_store.get(created['id'])
        assert record['source']=='test_pipeline' and record['session_id']
        assert (await http.get('/api/test-lab/runs/'+created['id']+'?user_name=someone-else')).status_code==404


@pytest.mark.asyncio
async def test_uploaded_existing_mock_curves_replace_the_preset_series_without_guessing(controller):
    transport=httpx.ASGITransport(app=app)
    async with httpx.AsyncClient(transport=transport,base_url='http://test') as http:
        ids=[]
        for name in ('mock_switch_states.json','mock_airflow_series.json'):
            data=(Path(__file__).parents[1]/'examples/main_fan'/name).read_bytes()
            reply=await http.post('/api/test-lab/assets',files={'file':(name,data,'application/json')})
            assert reply.status_code==201
            ids.append(reply.json()['id'])
        payload=request('abnormal',asset_ids=ids).model_dump(mode='json')
        response=await http.post('/api/test-lab/runs',json=payload)
        assert response.status_code==202,response.text
        await controller.run_job(business_store.claim('main_fan'))
        actual=(await http.get('/api/test-lab/runs/'+response.json()['id'])).json()
        assert actual['result']['platform_payload']['judgment']['resultCode']=='NORMAL'
        event=actual['result']['assessment']['switch_comparison']['events'][0]
        assert event['metrics']['flow_deviation_percent']==pytest.approx(5)
        assert event['metrics']['switch_interval_minutes']==6


@pytest.mark.asyncio
async def test_excel_download_and_preview_match_canonical_judgment(controller):
    value=service.start_run(request(),'lab-user')
    await controller.run_job(business_store.claim('main_fan'))
    transport=httpx.ASGITransport(app=app)
    async with httpx.AsyncClient(transport=transport,base_url='http://test') as http:
        run=(await http.get(f"/api/test-lab/runs/{value['id']}?user_name=lab-user")).json()
        item=run['artifacts'][0]
        preview=await http.get(item['preview_url']+'?user_name=lab-user')
        assert preview.status_code==200,preview.text
        workbook=preview.json()
        assert workbook['origin']=='generated'
        assert [s['name'] for s in workbook['sheets']]==['研判结果','性能曲线比较']
        rows=workbook['sheets'][0]['rows']
        assert any(row and row[0]=='额定转速（r/min）' for row in rows)
        assert rows[4][1]==run['result']['platform_payload']['judgment']['resultCode']=='NORMAL'
        assert rows[9][1]==6 and rows[10][1]==pytest.approx(5)
        downloaded=await http.get(item['url']+'?user_name=lab-user')
        assert downloaded.status_code==200 and downloaded.content.startswith(b'PK')
        assert 'application/vnd.openxmlformats-officedocument.spreadsheetml.sheet' in downloaded.headers['content-type']
        actual=service.artifact(run['id'],'result_excel','lab-user')
        with ZipFile(actual['path']) as archive:
            root=ET.fromstring(archive.read('xl/worksheets/sheet1.xml'))
            assert root.find(f".//{{{NS}}}c[@r='B5']/{{{NS}}}is/{{{NS}}}t").text=='NORMAL'
            assert float(root.find(f".//{{{NS}}}c[@r='B10']/{{{NS}}}v").text)==6
            assert root.findall(f'.//{{{NS}}}f')==[]
        assert (await http.get(item['url']+'?user_name=other')).status_code==404


def test_excel_preview_reads_cached_formulas_as_data_and_bounds_rows_and_columns(tmp_path):
    path=tmp_path/'external.xlsx'
    write_workbook(path,[{'name':'外部附件','rows':[['=HYPERLINK("https://invalid")',100],['后续',200],['第三行',300]],'widths':[30,20],'headers':[1]}])
    actual=preview_workbook(path,max_rows=2,max_cols=1)
    assert actual['sheets'][0]['rows']==[['=HYPERLINK("https://invalid")'],['后续']]
    assert actual['sheets'][0]['total_rows']==3 and actual['sheets'][0]['truncated']
    with ZipFile(path) as archive:
        assert ET.fromstring(archive.read('xl/worksheets/sheet1.xml')).findall(f'.//{{{NS}}}f')==[]


@pytest.mark.asyncio
async def test_actual_returned_excel_is_available_alongside_generated_export_and_cannot_escape_session(controller,tmp_path):
    from agents.workspace import session_root
    from runtime.business_models import ResultArtifact
    created=service.start_run(request(),'lab-user')
    await controller.run_job(business_store.claim('main_fan'))
    job=business_store.get(created['id'])
    result=json.loads(job['result_json'])
    folder=session_root('lab-user',job['session_id'])/'workspace/results'
    folder.mkdir(parents=True,exist_ok=True)
    write_workbook(folder/'actual.xlsx',[{'name':'实际返回数据','rows':[['风量（m³/min）'],[3800]],'headers':[1],'widths':[25]}])
    # 用接口契约模拟一个实际生成文件的业务工具；主通风机判识数据保持原样。
    result['artifacts']=[ResultArtifact(filename='实际返回.xlsx',path='/workspace/results/actual.xlsx').model_dump(),
        {'filename':'越界文件.xlsx','path':'/workspace/../../outside.xlsx'}]
    business_store.update(job['id'],result_json=json.dumps(result,ensure_ascii=False))
    record=service.public_run(job['id'],'lab-user')
    assert [a['origin'] for a in record['artifacts']]==['returned','generated']
    item=service.artifact(job['id'],record['artifacts'][0]['id'],'lab-user')
    assert preview_workbook(item['path'])['sheets'][0]['rows'][1][0]==3800
    with pytest.raises(ValueError):
        service.artifact(job['id'],record['artifacts'][0]['id'],'other-user')


@pytest.mark.asyncio
async def test_queue_cancel_only_cancels_the_selected_test(controller):
    first=service.start_run(request(),'lab-user')
    second=service.start_run(request(),'lab-user')
    await controller.cancel_job(first['id'])
    assert service.public_run(first['id'],'lab-user')['status']=='cancelled'
    assert service.public_run(second['id'],'lab-user')['status']=='queued'


@pytest.mark.asyncio
async def test_full_agent_mode_still_uses_the_existing_orchestrator(controller):
    from runtime.engine import ExecutionResult
    calls=[]
    async def execute(**kwargs):
        calls.append(kwargs)
        controlled=kwargs['graph_config']['configurable']
        result=await main_fan_service.analyze(controlled['business_task_input'],kwargs['user_name'],kwargs['session_id'],job_id=controlled['business_task_id'])
        return ExecutionResult('test-run',kwargs['session_id'],'success',result['result'],'',{},[],[],result)
    controller.executor=execute
    created=service.start_run(request(execution_mode='agent'),'lab-user')
    await controller.run_job(business_store.claim('main_fan'))
    assert len(calls)==1 and calls[0]['agent_name']=='main_fan'
    assert service.public_run(created['id'],'lab-user')['status']=='completed'


@pytest.mark.asyncio
async def test_workbench_can_be_disabled_without_changing_chat_or_business_api(controller,monkeypatch):
    monkeypatch.setenv('TEST_LAB_ENABLED','0')
    async with httpx.AsyncClient(transport=httpx.ASGITransport(app=app),base_url='http://test') as http:
        assert (await http.get('/api/test-lab/catalog')).status_code==404
        assert (await http.get('/api/business-agents')).status_code==200


@pytest.mark.parametrize('interval,before,after,direction,code',[
    (6,3800,3610,'1_to_2','NORMAL'), (6.5,3800,3344,'1_to_2','WARNING'),
    (10,3800,3420,'2_to_1','NORMAL'), (10.5,3800,3610,'2_to_1','WARNING'),
    (0,3800,3230,'1_to_2','ABNORMAL'), (120,3800,3040,'1_to_2','ABNORMAL'),
    (6,3800,4180,'1_to_2','NORMAL'), (6,3800,4370,'1_to_2','ABNORMAL'),
])
def test_three_input_generator_preserves_interval_flow_direction_and_boundaries(interval,before,after,direction,code):
    from algorithms.main_fan.models import ScreeningInput
    from algorithms.main_fan.equivalence import normalize_and_validate_fan_data, analyze_fan_switch_event
    from extensions.main_fan.pipeline import load_rules, run_pipeline
    settings=MainFanSimulation(switch_interval_minutes=interval,before_flow_m3_min=before,after_flow_m3_min=after,direction=direction)
    raw=build_input(settings)
    assert raw['reports']==[] and raw['use_gateway'] is False and raw['document_cache_policy']=='off'
    # 仅算法回归使用明确的合成报告，不会进入网页三项输入请求。
    raw['reports']=main_fan_input('normal')['reports']
    data=ScreeningInput.model_validate({k:v for k,v in raw.items() if k in ScreeningInput.model_fields})
    dataset=normalize_and_validate_fan_data(data,load_rules())
    switch=analyze_fan_switch_event(dataset)
    assert len(switch['events'])==1
    event=switch['events'][0]
    assert event['metrics']['switch_interval_minutes']==interval
    assert event['metrics']['flow_deviation_percent']==pytest.approx(abs(after-before)/before*100)
    assert event['from_fan_id']==('mock-report-fan-2' if direction=='2_to_1' else 'mock-report-fan-1')
    assessment,_,_=run_pipeline({'data':data.model_dump(mode='json'),
        'analysis':main_fan_service._screen(data).model_dump(mode='json'),
        'policy':load_rules().model_dump(mode='json')})
    assert assessment['decision']['result_code']==code


@pytest.mark.asyncio
async def test_three_inputs_always_start_fresh_agent_and_workspace_without_document_cache(controller,monkeypatch):
    from agents.workspace import session_root
    from algorithms.main_fan.models import DocumentFacts
    from db import chat_store
    from runtime.engine import ExecutionResult
    http=httpx.AsyncClient(transport=httpx.ASGITransport(app=app),base_url='http://test')
    upload=await http.post('/api/test-lab/assets',files={'file':('双机报告.pdf',b'%PDF-1.4\nunit test fixture','application/pdf')})
    asset=upload.json()['id']
    calls=[]
    async def read(path,virtual,**kwargs):
        calls.append(kwargs)
        return DocumentFacts(reports=main_fan_input('normal')['reports'],extraction_audit={'cache_used':False,'cache_policy':kwargs['cache_policy']})
    monkeypatch.setattr(main_fan_service,'read_document',read)
    seen=[]
    async def execute(**kwargs):
        root=session_root(kwargs['user_name'],kwargs['session_id'])
        assert not (root/'workspace/previous-run.txt').exists()
        assert chat_store.get_messages(kwargs['session_id'])==[]
        controlled=kwargs['graph_config']['configurable']
        assert controlled['business_task_input']['reports']==[]
        seen.append(root)
        result=await main_fan_service.analyze(controlled['business_task_input'],kwargs['user_name'],kwargs['session_id'],job_id=controlled['business_task_id'])
        (root/'workspace/previous-run.txt').write_text('仅供本次工作区使用')
        return ExecutionResult('unit-test',kwargs['session_id'],'success',result['result'],'',{},[],[],result)
    controller.executor=execute
    records=[]
    for after,code in [(3610,'NORMAL'),(3344,'WARNING')]:
        response=await http.post('/api/test-lab/runs',json={
            'agent_name':'main_fan','scenario_id':'three_inputs','label':'三项输入', 'input':{},
            'asset_ids':[asset],'execution_mode':'agent',
            'main_fan_simulation':{'after_flow_m3_min':after}})
        assert response.status_code==202,response.text
        created=response.json()
        await controller.run_job(business_store.claim('main_fan'))
        actual=service.public_run(created['id'],'default')
        assert actual['status']=='completed',actual['error']
        assert actual['result']['platform_payload']['judgment']['resultCode']==code
        assert actual['request']['isolation']['history_reused'] is False
        assert actual['request']['input']['document_cache_policy']=='off'
        assert (seen[-1]/'workspace/test_inputs/主备风机开停曲线.json').is_file()
        records.append(actual)
    assert records[0]['session_id']!=records[1]['session_id'] and seen[0]!=seen[1]
    assert len(calls)==2 and all(c['cache_policy']=='off' for c in calls)
    assert db_conn.scalar('SELECT COUNT(*) FROM main_fan_documents',())==0
    await http.aclose()


@pytest.mark.asyncio
async def test_three_inputs_require_pdf_full_agent_and_refuse_hidden_report_data(controller):
    async with httpx.AsyncClient(transport=httpx.ASGITransport(app=app),base_url='http://test') as http:
        body={'agent_name':'main_fan','input':{},'execution_mode':'agent','main_fan_simulation':{}}
        missing=await http.post('/api/test-lab/runs',json=body)
        assert missing.status_code==400 and 'PDF' in missing.text
        body['execution_mode']='pipeline'
        wrong_mode=await http.post('/api/test-lab/runs',json=body)
        assert wrong_mode.status_code==400 and '完整智能体' in wrong_mode.text
        body['execution_mode']='agent'
        body['input']={'reports':main_fan_input('normal')['reports']}
        injected=await http.post('/api/test-lab/runs',json=body)
        assert injected.status_code==400 and '混入' in injected.text
        body['input']={}
        body['main_fan_simulation']={'switch_source':'file'}
        invalid=await http.post('/api/test-lab/runs',json=body)
        assert invalid.status_code==422


@pytest.mark.asyncio
async def test_three_input_slots_replace_entire_curve_inputs_and_reject_swapped_files(controller):
    async with httpx.AsyncClient(transport=httpx.ASGITransport(app=app),base_url='http://test') as http:
        pdf=(await http.post('/api/test-lab/assets',files={'file':('双机.pdf',b'%PDF-1.4\nunit test fixture','application/pdf')})).json()['id']
        curves=[]
        for name in ('mock_switch_states.json','mock_airflow_series.json'):
            contents=(Path(__file__).parents[1]/'examples/main_fan'/name).read_bytes()
            uploaded=await http.post('/api/test-lab/assets',files={'file':(name,contents,'application/json')})
            curves.append(uploaded.json()['id'])
        body={'agent_name':'main_fan','input':{},'execution_mode':'agent','asset_ids':[pdf],
            'main_fan_simulation':{'switch_source':'file','flow_source':'file',
                'switch_asset_id':curves[0],'flow_asset_id':curves[1]}}
        response=await http.post('/api/test-lab/runs',json=body)
        assert response.status_code==202,response.text
        actual=response.json()['request']
        assert len(actual['assets'])==3 and actual['input']['reports']==[]
        for key in ('fan_state_records','operating_series'):
            assert actual['input'][key]
            assert all(row['evidence']['source'].startswith('test-lab://assets/') for row in actual['input'][key])
        body['main_fan_simulation']['switch_asset_id'],body['main_fan_simulation']['flow_asset_id']=curves[1],curves[0]
        swapped=await http.post('/api/test-lab/runs',json=body)
        assert swapped.status_code==400 and '对应的时序' in swapped.text