vent_agent_2.0/tests/test_main_fan_targets.py

"""简短业务入口：身份解析、环境隔离、网关默认策略与未决身份阻断。"""
import json
from pathlib import Path
from types import SimpleNamespace

import pytest

from extensions.main_fan.schemas import MainFanInput, MainFanResult
from extensions.main_fan.targets import load_catalog, TargetResolutionError
from extensions.main_fan.tools import analyze_main_fan


@pytest.fixture
def target_catalog(monkeypatch):
    root=Path(__file__).resolve().parents[1]
    monkeypatch.setenv('MAIN_FAN_TARGETS_FILE',str(root/'examples/main_fan/mock_targets.json'))
    monkeypatch.setenv('SHANXI_GATEWAY_URL','http://127.0.0.1:8766')
    monkeypatch.setenv('SHANXI_GATEWAY_TOKEN','synthetic-test-token')
    monkeypatch.setenv('SHANXI_GATEWAY_ENABLED','1')
    monkeypatch.setenv('MAIN_FAN_DEFAULT_DOCUMENT_CACHE_POLICY','refresh')


@pytest.mark.parametrize('value',[
    {'target':'众源煤业1号主通风机'},
    {'target':'mock-report-fan-1'},
    '{"target":"众源煤业1号主通风机"}',
    '审核众源煤业1号主通风机',
    '请对陕西众源煤业集团有限公司的一号主通风机进行研判',
    {'mine_name':'众源煤业','fan_name':'1#通风机'},
    {'fan_id':'mock-report-fan-1'},
    {'fanId':'mock-report-fan-1'},
])
def test_single_target_resolves_configured_identity_and_internal_defaults(target_catalog,value):
    result=MainFanInput.model_validate(value)
    assert result.mine_code=='mock-zy'
    assert result.mine_name=='陕西众源煤业集团有限公司'
    assert result.fan_id=='mock-report-fan-1'
    assert result.fan_name=='1号通风机'
    assert result.use_gateway is True and result.document_cache_policy=='refresh'
    assert result.reports==[] and result.points==[] and result.confirmed_bindings==[]


def test_second_fan_is_not_replaced_with_first_fan(target_catalog):
    result=MainFanInput.model_validate({'target':'审核众源煤业二号主通风机'})
    assert result.fan_id=='mock-report-fan-2' and result.fan_name=='2号通风机'


@pytest.mark.parametrize('value',[
    {'target':'众源煤业3号主通风机'},
    {'target':'其他煤矿1号主通风机'},
    {'target':'众源煤业1号主通风机','mine_code':'other-mine'},
    {'target':'众源煤业1号主通风机','mine_name':'其他矿井'},
    {'target':'众源煤业1号主通风机','fan_id':'mock-report-fan-2'},
    {'target':'众源煤业1号主通风机','fan_name':'2号通风机'},
    {'target':'众源煤业1号主通风机','mine_code':123},
    {'target':'众源煤业1号主通风机','mine_code':'mock-zy','mineCode':'other'},
    '{"target":',
])
def test_ambiguous_unknown_or_conflicting_identity_is_rejected(target_catalog,value):
    with pytest.raises(ValueError):MainFanInput.model_validate(value)


def test_mock_identity_catalog_cannot_be_used_with_real_gateway(target_catalog,monkeypatch):
    monkeypatch.setenv('SHANXI_GATEWAY_URL','https://real-gateway.invalid')
    with pytest.raises(ValueError,match='绑定的网关'):load_catalog()
    with pytest.raises(ValueError,match='绑定的网关'):
        MainFanInput.model_validate({'target':'众源煤业1号主通风机'})


def test_target_resolution_without_catalog_never_fabricates_codes(monkeypatch):
    monkeypatch.delenv('MAIN_FAN_TARGETS_FILE',raising=False)
    with pytest.raises(ValueError):MainFanInput.model_validate({'target':'众源煤业1号主通风机'})


def test_explicit_local_material_and_cache_choices_are_preserved(target_catalog):
    result=MainFanInput.model_validate({'target':'众源煤业1号主通风机',
        'materials':[{'path':'/workspace/uploads/raw.pdf'}],'document_cache_policy':'off'})
    assert result.use_gateway is False and result.document_cache_policy=='off'
    explicit=MainFanInput.model_validate({'target':'众源煤业1号主通风机',
        'use_gateway':False,'document_cache_policy':'auto'})
    assert explicit.use_gateway is False and explicit.document_cache_policy=='auto'


def test_disabling_gateway_does_not_activate_mock_implicitly(target_catalog,monkeypatch):
    monkeypatch.setenv('SHANXI_GATEWAY_ENABLED','0')
    assert MainFanInput.model_validate({'target':'众源煤业1号主通风机'}).use_gateway is False


def test_complete_input_and_camel_case_remain_compatible(monkeypatch):
    monkeypatch.delenv('MAIN_FAN_TARGETS_FILE',raising=False)
    value=MainFanInput.model_validate({'mineCode':'formal-code','mineName':'正式矿井',
        'fanId':'formal-fan','fan_name':'1号通风机','use_gateway':False})
    assert value.mine_code=='formal-code' and value.fan_id=='formal-fan'
    assert value.document_cache_policy=='auto'


@pytest.mark.asyncio
async def test_short_tool_call_passes_server_resolved_parameters(target_catalog,monkeypatch):
    seen=[]
    async def fake_analyze(data,user,session,**kwargs):
        seen.append((data,user,session))
        return MainFanResult(status='needs_materials',mine_code=data['mine_code'],mine_name=data['mine_name'],
            check_object=data['fan_name'],result_title='测试',result='仅验证输入路由').model_dump(mode='json')
    monkeypatch.setattr('extensions.main_fan.service.analyze',fake_analyze)
    runtime=SimpleNamespace(config={'configurable':{'workspace_user':'test','workspace_session':'test-session'}})
    await analyze_main_fan('审核众源煤业1号主通风机',runtime)
    data,user,session=seen[0]
    assert data['fan_id']=='mock-report-fan-1' and data['use_gateway'] is True
    assert data['document_cache_policy']=='refresh' and not data.get('reports')
    assert (user,session)==('test','test-session')


@pytest.mark.asyncio
async def test_ambiguous_tool_target_never_calls_gateway_or_analyzer(target_catalog,monkeypatch):
    from extensions.main_fan.targets import TargetCatalog
    content=load_catalog().model_dump()
    for record in content['targets']:record.pop('fan_pair')
    monkeypatch.setattr('extensions.main_fan.targets.load_catalog',lambda:TargetCatalog.model_validate(content))
    async def forbidden(*args,**kwargs):raise AssertionError('身份未定不得分析')
    monkeypatch.setattr('extensions.main_fan.service.analyze',forbidden)
    runtime=SimpleNamespace(config={'configurable':{'workspace_user':'test','workspace_session':'test-session'}})
    result=await analyze_main_fan('审核众源煤业主通风机',runtime)
    assert result['status']=='needs_materials' and not result['analysis_id']
    assert '1号通风机' in result['result'] and '2号通风机' in result['result']


@pytest.mark.parametrize('value',[{'target':'众源煤业'},{'mine_code':'mock-zy'},'审核众源煤业主通风机','审核众源煤业主通风机及备用是否一致'])
def test_mine_request_resolves_one_configured_fan_pair(target_catalog,value):
    result=MainFanInput.model_validate(value)
    assert result.fan_pair==['mock-report-fan-1','mock-report-fan-2']
    assert result.paired_assessment is True


def test_job_api_accepts_one_target_parameter(target_catalog,monkeypatch):
    from runtime.business_tasks import submit
    from db import business_store
    monkeypatch.setenv('SHANXI_GATEWAY_ENABLED','0')
    job=submit('main_fan',{'target':'众源煤业1号主通风机'},user_name='one-parameter',publish=False)
    stored=json.loads(business_store.get(job['id'])['payload'])['input']
    assert stored['mine_code']=='mock-zy' and stored['fan_id']=='mock-report-fan-1'
    assert stored['document_cache_policy']=='refresh'