vent_agent_2.0/tests/test_main_fan_operating_mock.py

"""用户指定的10分钟/10%规则：边界、独立证据、网关取数和整体报告。"""
from datetime import datetime, timedelta
from pathlib import Path
import base64
import json
import re

import httpx
import pytest

from algorithms.main_fan.equivalence import analyze_fan_switch_event, normalize_and_validate_fan_data
from algorithms.main_fan.models import ScreeningInput
from extensions.main_fan import data_source, service
from extensions.main_fan.pipeline import load_rules
from extensions.shaanxi.client import GatewayClient
from extensions.shaanxi.settings import GatewaySettings
from mock_gateway.main_fan import INTERNAL_TOKEN, MockState, create_app
from mock_gateway.operating_data import load_operating_data

EXAMPLES = Path(__file__).resolve().parents[1]/'examples/main_fan'


def sample(*, gap=6, after_flow=3610):
    source=load_operating_data(EXAMPLES)
    stop=datetime.fromisoformat('2026-10-14T10:00:00+08:00')
    start=stop+timedelta(minutes=gap)
    for record in source['fan_state_records']:
        if record['fan_id']=='mock-report-fan-2':
            record['state']='running' if datetime.fromisoformat(record['observed_at'])>=start else 'stopped'
    for series in source['operating_series']:
        if series['fan_id']=='mock-report-fan-2':
            for point in series['samples']:
                at=datetime.fromisoformat(point['observed_at'])
                point['value']=0 if at<start else after_flow if at>=start+timedelta(minutes=2) else after_flow/2
    return ScreeningInput(mine_code='mock-zy',mine_name='陕西众源煤业集团有限公司',
        fan_pair=['mock-report-fan-1','mock-report-fan-2'],evaluation_time='2026-10-14T11:00:00+08:00',
        **{k:source[k] for k in ('devices','fan_state_records','operating_series')})


def analyze(data):
    return analyze_fan_switch_event(normalize_and_validate_fan_data(data,load_rules()))


@pytest.mark.parametrize('gap,flow,checks,status',[
    (6,3610,('pass','pass'),'pass'),
    (10,3420,('pass','pass'),'pass'),
    (10,4180,('pass','pass'),'pass'),
    (11,3610,('fail','pass'),'fail'),
    (6,3040,('pass','fail'),'fail'),
    (6,4560,('pass','fail'),'fail'),
    (11,3040,('fail','fail'),'fail'),
    (0,3800,('pass','pass'),'pass'),
    (-2,3800,('pass','pass'),'pass'),
    (6,0,('pass','fail'),'fail'),
])
def test_switch_interval_and_flow_change_are_independent_and_inclusive(gap,flow,checks,status):
    result=analyze(sample(gap=gap,after_flow=flow))
    assert result['status']==status and len(result['events'])==1
    event=result['events'][0]
    assert event['checks']==dict(zip(('switch_interval','flow_change'),checks))
    assert event['timing']['interval_minutes']==max(gap,0)
    assert event['timing']['overlap_minutes']==max(-gap,0)
    assert event['metrics']['flow_deviation_percent']==pytest.approx(abs(flow-3800)/3800*100)
    assert event['changes']['flow']['before']*60==pytest.approx(3800)
    assert event['changes']['flow']['after']*60==pytest.approx(flow)
    assert event['before_window']['flow']['sample_count']==30
    assert event['after_window']['flow']['sample_count']==30
    assert 'pressure' not in event['assessed_metrics']


@pytest.mark.parametrize('kind',['unconfirmed','duplicate_sensor','missing_samples','conflicting_samples','zero_baseline'])
def test_timing_is_still_reported_when_flow_cannot_be_compared(kind):
    data=sample()
    if kind=='unconfirmed':data.operating_series[0].mapping_confirmed=False
    elif kind=='duplicate_sensor':data.operating_series.append(data.operating_series[0].model_copy(deep=True))
    elif kind=='missing_samples':data.operating_series[0].samples=[]
    elif kind=='conflicting_samples':
        series=data.operating_series[0]
        series.samples.extend(p.model_copy(update={'value':p.value+30}) for p in list(series.samples))
    elif kind=='zero_baseline':
        for p in data.operating_series[0].samples:p.value=0
    result=analyze(data)
    assert result['status']=='insufficient'
    event=result['events'][0]
    assert event['checks']=={'switch_interval':'pass','flow_change':'insufficient'}
    assert event['timing']['interval_minutes']==6
    assert 'flow' not in event['changes']


@pytest.mark.parametrize('kind',['stage_only','unknown_state','conflict','missing_transition','foreign','future'])
def test_state_chain_must_be_whole_fan_continuous_and_in_scope(kind):
    data=sample()
    if kind=='stage_only':
        for r in data.fan_state_records:r.complete_fan_state=False
    elif kind=='unknown_state':
        for r in data.fan_state_records:
            if r.observed_at.minute==3 and r.observed_at.hour==10:r.state='unknown'
    elif kind=='conflict':
        r=next(r for r in data.fan_state_records if r.fan_id.endswith('-1') and r.observed_at.hour==10 and r.observed_at.minute==0)
        data.fan_state_records.append(r.model_copy(update={'state':'running'}))
    elif kind=='missing_transition':
        data.fan_state_records=[r for r in data.fan_state_records if not (r.observed_at.hour==10 and r.observed_at.minute<=7)]
    elif kind=='foreign':
        for r in data.fan_state_records:r.mine_code='another-mine'
    else:data.evaluation_time=datetime.fromisoformat('2026-10-14T09:30:00+08:00')
    result=analyze(data)
    assert result['events']==[] and result['status']=='insufficient'


def test_future_samples_cannot_fill_after_window():
    data=sample()
    data.evaluation_time=datetime.fromisoformat('2026-10-14T10:10:00+08:00')
    event=analyze(data)['events'][0]
    assert event['timing']['interval_minutes']==6
    assert not event['changes']


@pytest.mark.asyncio
async def test_mock_gateway_forwards_both_curves_and_filters_mine(tmp_path,monkeypatch):
    (tmp_path/'original.pdf').write_bytes(b'original mock bytes')
    state=MockState.from_raw_files(tmp_path,operating_data=load_operating_data(EXAMPLES))
    transport=httpx.ASGITransport(app=create_app(state))
    client=GatewayClient(GatewaySettings('http://mock',INTERNAL_TOKEN,'http://mock'),transport=transport)
    for key,path in {'MAIN_FAN_DEVICES_PATH':'/mock-platform/main-fan/devices','MAIN_FAN_SERIES_PATH':'/mock-platform/main-fan/series',
                     'MAIN_FAN_OPERATION_LOG_PATH':'/mock-platform/main-fan/states','MAIN_FAN_POINTS_PATH':'',
                     'MAIN_FAN_REPORTS_PATH':'','MAIN_FAN_SWITCHES_PATH':''}.items():monkeypatch.setenv(key,path)
    try:
        await client.call('POST','/internal/agents/register','mock-main-fan',json={'agentCode':'mock-main-fan','checkTypeCode':'mock-fan-screening'})
        incoming=await data_source.fetch(client,'mock-main-fan','mock-zy')
        assert len(incoming['devices'])==2 and len(incoming['operating_series'])==2 and len(incoming['fan_state_records'])==202
        data=ScreeningInput.model_validate({**sample().model_dump(mode='json'),
            **{k:incoming[k] for k in ('devices','operating_series','fan_state_records')}})
        assert analyze(data)['status']=='pass'
        unrelated=await data_source.fetch(client,'mock-main-fan','mock-ci')
        assert all(not rows for rows in unrelated.values())
        async with httpx.AsyncClient(transport=transport,base_url='http://mock') as http:
            raw=(await http.get('/mock/main-fan/curves/switch-states')).json()
            assert raw['data_kind']=='synthetic' and len(raw['series'])==2
        assert all(s['evidence']['method']=='synthetic' for s in incoming['operating_series'])
        assert 'document_facts' not in state.sample
    finally:await client.close()


@pytest.mark.asyncio
async def test_report_explains_both_rules_and_labels_mock_curves():
    result=await service.analyze(sample().model_dump(mode='json'),'curve-u','curve-s')
    text=result['result']
    assert result['status']=='completed'
    assert '6.00分钟' in text and '5.00%' in text
    assert '不超过10分钟' in text and '不超过10%' in text
    assert '3800.00' in text and '3610.00' in text
    assert '模拟' in text and '需复核' not in text
    assert '风压' not in text
    assert len(result['sources'])==2
    encoded=re.search(r'!\[主备风机开停与风量时序曲线\]\(data:image/svg\+xml;base64,([^)]*)\)',text).group(1)
    svg=base64.b64decode(encoded).decode()
    assert '主备风机开停曲线' in svg and '主备风机风量曲线' in svg
    assert all(t in svg for t in ('10:00:00','10:06:00','模拟运行时序'))
    assert result['assessment']['switch_comparison']['status']=='pass'


def test_operating_fixture_identity_mismatch_is_rejected(tmp_path):
    for name in ('mock_switch_states.json','mock_airflow_series.json'):
        raw=json.loads((EXAMPLES/name).read_text())
        if name=='mock_airflow_series.json':raw['mine_code']='another'
        (tmp_path/name).write_text(json.dumps(raw))
    with pytest.raises(ValueError,match='矿井身份不一致'):load_operating_data(tmp_path)