vent_agent_2.0/tests/test_result_platform.py

"""结果平台独立性、业务编码、固定报文重试和 SQLite 幂等接收。"""
from datetime import date, datetime
import json

import httpx
import pytest

from algorithms.main_fan.models import Evidence, PerformancePoint, PerformanceReport
from db import business_store as store, db_conn
from extensions.main_fan import service
from extensions.main_fan.result_output import report_text
from extensions.result_platform.client import ResultPlatformClient, ResultPlatformError
from extensions.result_platform.settings import ResultPlatformSettings
from mock_gateway.operating_data import load_operating_data
from mock_result_platform.app import LOCAL_TOKEN, ResultStore, create_app
from runtime.business_tasks import BusinessTaskController, submit
from runtime.engine import ExecutionResult
from runtime.result_models import PlatformResultPayload
from pydantic import ValidationError


def complete_data(*, after_flow=None, angle=35):
    # 完整的测试输入；不依赖外部模型、原件缓存或既有研判结论。
    incoming = load_operating_data(db_conn.config.ROOT / "examples/main_fan")
    data = dict(mine_code="mock-zy", mine_name="陕西众源煤业集团有限公司",
        fan_id="mock-report-fan-1", fan_name="1号通风机", fan_pair=["mock-report-fan-1", "mock-report-fan-2"],
        evaluation_time="2026-10-14T11:00:00+08:00", use_gateway=False,
        **{key: incoming[key] for key in ("devices", "fan_state_records", "operating_series")})
    for device in data["devices"]:
        device.update(fan_role="standby" if device["device_code"].endswith("-1") else "in_service",
                      availability_status="available", observed_at=data["evaluation_time"])
    reports = []
    for number in (1, 2):
        evidence = Evidence(source=f"/workspace/fan-{number}.pdf", method="synthetic", page=1)
        points = [PerformancePoint(label="运行工况" if index == 1 else f"测点{index}",
            frequency_hz=50, flow_m3_s=q, static_pressure_pa=500, shaft_power_kw=80,
            reported_efficiency_pct=q*500/800, evidence=evidence) for index, q in enumerate((40, 50, 60, 70))]
        reports.append(PerformanceReport(report_id=f"R-{number}", mine_code=data["mine_code"], mine_name=data["mine_name"],
            fan_name=f"{number}号通风机", model="FBCDZ-No24", rated_power_raw="2×185kW", blade_angle_deg=angle,
            inspected_on=date(2026, 10, 14), evidence=evidence, points=points).model_dump(mode="json"))
    data["reports"] = reports
    if after_flow is not None:
        for series in data["operating_series"]:
            if series["fan_id"].endswith("-2"):
                for point in series["samples"]:
                    if point["value"] > 3000:
                        point["value"] = after_flow
    return data


def settings(url="http://results.test/api/judgments", **kwargs):
    return ResultPlatformSettings(url=url, token=LOCAL_TOKEN, **kwargs)


async def result_payload():
    result = await service.analyze(complete_data(), "test", "session")
    return result["platform_payload"]


@pytest.mark.asyncio
async def test_complete_case_emits_the_agreed_normal_schema_and_whole_report():
    result = await service.analyze(complete_data(), "normal", "session")
    payload = result["platform_payload"]
    assert set(payload) == {"agentCode", "agentName", "taskId", "checkedAt", "mineCode", "mineName", "objectDetail", "dataDetail", "judgment"}
    assert set(payload["judgment"]) == {"resultCode", "result", "reason", "evidenceLink"}
    assert payload["judgment"]["resultCode"] == "NORMAL"
    assert "NORMAL" in result["result"]
    assert all(name in payload["objectDetail"] for name in ("1号通风机", "2号通风机"))
    assert "6分钟" in payload["judgment"]["result"] and "5.00%" in payload["judgment"]["result"]
    assert "3610" in payload["dataDetail"] and "3800" in payload["dataDetail"]
    assert payload["judgment"]["reason"] == report_text(result["result"])
    assert "data:image" not in payload["judgment"]["reason"]
    assert payload["judgment"]["evidenceLink"] == ""
    assert all(isinstance(value, str) for key, value in payload.items() if key != "judgment")
    assert all(isinstance(value, str) for value in payload["judgment"].values())
    restored = service.validate_result({**result, "platform_payload": {**payload, "judgment": {**payload["judgment"], "resultCode": "ABNORMAL"}}}, "normal", "session")
    assert restored == result


@pytest.mark.asyncio
async def test_four_codes_are_the_entire_public_contract_and_leaf_values_are_strings():
    payload = await result_payload()
    for code in ("NORMAL", "WARNING", "ABNORMAL", "UNDETERMINED"):
        assert PlatformResultPayload.model_validate({**payload, "judgment": {**payload["judgment"], "resultCode": code}}).judgment.resultCode == code
    for code in ("PARTIAL", "EXEMPTED", "Warning"):
        with pytest.raises(ValidationError):
            PlatformResultPayload.model_validate({**payload, "judgment": {**payload["judgment"], "resultCode": code}})
    with pytest.raises(ValidationError):
        PlatformResultPayload.model_validate({**payload, "mineCode": 123})


@pytest.mark.asyncio
async def test_start_does_not_register_or_subscribe_to_data_gateway():
    class DataOnlyGateway:
        def __getattr__(self, name):
            raise AssertionError(f"启动时不应调用取数网关：{name}")
    controller = BusinessTaskController(DataOnlyGateway())
    await controller.start()
    try:
        assert not any(task.get_name().startswith("business-gateway") for task in controller.tasks)
    finally:
        await controller.close()


@pytest.mark.asyncio
@pytest.mark.parametrize("change,code", [("no_angle", "NORMAL"), ("power", "WARNING"),
    ("flow", "ABNORMAL"), ("flow_warning", "WARNING"), ("severe_boundary", "ABNORMAL"), ("boundary", "NORMAL")])
async def test_code_comes_from_actual_assessment_and_preserves_inclusive_boundaries(change, code):
    data = complete_data()
    if change == "no_angle":
        for report in data["reports"]:
            report["blade_angle_deg"] = None
    elif change == "power":
        data["reports"][1]["rated_power_raw"] = "2×200kW"
    elif change in {"flow", "flow_warning", "severe_boundary", "boundary"}:
        data = complete_data(after_flow={"flow":3040, "flow_warning":3382, "severe_boundary":3230, "boundary":3420}[change])
    result = await service.analyze(data, "code", "session")
    assert result["platform_payload"]["judgment"]["resultCode"] == code
    assert code in result["result"]
    assert "PARTIAL" not in json.dumps(result["platform_payload"])
    assert "EXEMPTED" not in json.dumps(result["platform_payload"])


@pytest.mark.asyncio
async def test_three_passes_are_normal_despite_unknown_metadata_ocr_and_table_graph_difference(monkeypatch):
    data = complete_data(angle=None)
    for device in data["devices"]:
        device.update(fan_role="unknown", availability_status="unknown", observed_at=None)
    point = data["reports"][0]["points"][1]
    data["reports"][0]["curves"] = [{**point, "label":"性能曲线", "static_pressure_pa":480,
                                      "reported_efficiency_pct":30}]
    original = service.parse_fan_evidence_documents
    async def with_ocr_notes(*args, **kwargs):
        parsed = await original(*args, **kwargs)
        parsed[1].extend(f"PDF页{page}：封面或声明页，未载明叶片角度。" for page in range(1,27))
        return parsed
    monkeypatch.setattr(service, "parse_fan_evidence_documents", with_ocr_notes)
    result = await service.analyze(data, "three-pass", "session")
    assert result["assessment"]["decision"]["items"] == dict(parameters="pass", switch_interval="pass", flow_change="pass")
    assert result["assessment"]["decision"]["result_code"] == "NORMAL"
    assert result["platform_payload"]["judgment"]["resultCode"] == "NORMAL"
    assert any(f["code"] == "document_extraction" for f in result["analysis"]["findings"])
    assert any(f["code"] == "curve_table_disagreement" for f in result["analysis"]["findings"])
    assert "检测资料一致性事项" in result["result"] and "相差20.00 Pa" in result["result"]
    assert "PDF页" not in result["result"] and "需复核" not in result["result"]


@pytest.mark.asyncio
async def test_supplementary_curve_and_pressure_failures_cannot_veto_three_passes():
    data = complete_data()
    for point in data["reports"][1]["points"]:
        point.update(static_pressure_pa=900, reported_efficiency_pct=point["flow_m3_s"]*900/800)
    for series in list(data["operating_series"]):
        data["operating_series"].append({**series, "metric":"static_pressure", "unit":"Pa",
            "sensor_id":series["sensor_id"]+"-pressure", "samples":[{**p, "value":500 if series["fan_id"].endswith("-1") else 1000} for p in series["samples"]]})
    result = await service.analyze(data, "supplementary", "session")
    assessment = result["assessment"]
    assert assessment["performance_comparison"]["status"] == "fail"
    assert assessment["switch_comparison"]["status"] == "fail"
    assert assessment["switch_comparison"]["events"][0]["checks"] == dict(switch_interval="pass", flow_change="pass")
    assert result["platform_payload"]["judgment"]["resultCode"] == "NORMAL"


@pytest.mark.asyncio
@pytest.mark.parametrize("change", ["other_parameter", "long_interval"])
async def test_clear_minor_issue_is_warning_instead_of_abnormal(change):
    data = complete_data()
    if change == "other_parameter":
        for number, report in enumerate(data["reports"]):
            report["motor_model"] = f"MOTOR-{number}"
    else:
        start = datetime.fromisoformat("2026-10-14T10:11:00+08:00")
        for record in data["fan_state_records"]:
            if record["fan_id"].endswith("-2"):
                record["state"] = "running" if datetime.fromisoformat(record["observed_at"]) >= start else "stopped"
    result = await service.analyze(data, "minor-issue", "session")
    assert result["platform_payload"]["judgment"]["resultCode"] == "WARNING"
    assert result["assessment"]["decision"]["severe_findings"] == []


@pytest.mark.asyncio
@pytest.mark.parametrize("change", ["unconfirmed", "missing", "conflicting", "missing_parameters"])
async def test_invalid_core_data_is_undetermined_and_not_assumed_to_pass(change):
    data = complete_data()
    if change == "unconfirmed":
        data["operating_series"][0]["mapping_confirmed"] = False
    elif change == "missing":
        data["operating_series"][0]["samples"] = []
    elif change == "conflicting":
        series = data["operating_series"][0]
        series["samples"].extend({**p,"value":p["value"]+30} for p in list(series["samples"]))
    else:
        for report in data["reports"]:
            report["model"] = ""
    result = await service.analyze(data, "bad-core", "session")
    assert result["platform_payload"]["judgment"]["resultCode"] == "UNDETERMINED"
    assert "insufficient" in result["assessment"]["decision"]["items"].values()


@pytest.mark.asyncio
async def test_partial_scope_is_not_normal_and_does_not_list_absent_projects():
    data = complete_data()
    data.update(fan_state_records=[], operating_series=[], devices=[])
    result = await service.analyze(data, "scope", "session")
    payload = result["platform_payload"]
    assert payload["judgment"]["resultCode"] == "UNDETERMINED"
    assert "切换" not in payload["judgment"]["reason"]
    assert "备用状态" not in payload["dataDetail"]
    assert "开停" not in payload["dataDetail"]


def test_report_tables_are_independent_even_after_operating_point_paragraphs():
    text = report_text("## 比较\n| 设备 | 风量 |\n| --- | ---: |\n| 1号 | 3800 |\n\n说明\n\n| 指标 | 偏差 |\n| --- | --- |\n| 静压 | 2% |\n![图](data:image/svg+xml;base64,abc)")
    assert "设备：1号；风量：3800" in text and "指标：静压；偏差：2%" in text
    assert "base64" not in text


@pytest.mark.asyncio
async def test_multiple_frequency_operating_points_keep_headers_and_both_fans_in_export():
    data = complete_data()
    for report in data["reports"]:
        points = list(report["points"])
        report["points"] = [{**point, "frequency_hz": frequency} for frequency in (30, 40, 50) for point in points]
    result = await service.analyze(data, "multi-frequency", "session")
    reason = result["platform_payload"]["judgment"]["reason"]
    for frequency in (30, 40, 50):
        for number in (1, 2):
            assert f"频率与设备：{frequency} Hz {number}号通风机；风量" in reason


@pytest.mark.asyncio
async def test_local_result_platform_persists_deduplicates_and_rejects_changed_task(tmp_path):
    payload = await result_payload()
    storage = ResultStore(tmp_path / "received.sqlite")
    async with httpx.AsyncClient(transport=httpx.ASGITransport(app=create_app(storage)), base_url="http://results.test") as http:
        assert (await http.post("/api/judgments", json=payload)).status_code == 401
        headers = {"Authorization": f"Bearer {LOCAL_TOKEN}", "Idempotency-Key": payload["taskId"]}
        first = await http.post("/api/judgments", headers=headers, json=payload)
        again = await http.post("/api/judgments", headers=headers, json=payload)
        assert first.status_code == 200 and first.json() == again.json()
        altered = {**payload, "dataDetail": "changed"}
        assert (await http.post("/api/judgments", headers=headers, json=altered)).status_code == 409
        assert storage.count() == 1
        invalid = {**payload, "judgment": {**payload["judgment"], "resultCode": "PARTIAL"}}
        assert (await http.post("/api/judgments", headers=headers, json=invalid)).status_code == 422
    reopened = ResultStore(storage.path)
    assert reopened.count() == 1 and reopened.rows()[0]["payload"] == payload


@pytest.mark.asyncio
async def test_retries_read_frozen_payload_after_controller_restart_and_do_not_touch_gateway(tmp_path):
    storage = ResultStore(tmp_path / "received.sqlite")
    app = create_app(storage, fail_first=1)
    publisher = ResultPlatformClient(settings(), transport=httpx.ASGITransport(app=app))
    calls = []
    class DataGateway:
        def __getattr__(self, name):
            raise AssertionError(f"本地资料研判和结果推送不应调用网关：{name}")
    async def executor(**kwargs):
        calls.append(kwargs)
        options = kwargs["graph_config"]["configurable"]
        result = await service.analyze(options["business_task_input"], kwargs["user_name"], kwargs["session_id"], job_id=options["business_task_id"])
        result["platform_payload"]["judgment"]["resultCode"] = "ABNORMAL"
        return ExecutionResult("run", kwargs["session_id"], "success", "", "", {}, [], [], result)
    controller = BusinessTaskController(DataGateway(), publisher=publisher, executor=executor)
    try:
        job, _ = store.enqueue("main_fan", "retry", complete_data(), [], publish=True)
        await controller.run_job(store.claim("main_fan"))
        completed = store.get(job["id"])
        before = completed["push_payload"]
        assert completed["push_status"] == "pending"
        assert json.loads(before)["judgment"]["resultCode"] == "NORMAL"
        await controller.publish_job(completed)
        failed = store.get(job["id"])
        assert failed["push_status"] == "pending" and failed["push_attempts"] == 1
        restored = BusinessTaskController(publisher=publisher, executor=executor)
        await restored.publish_job(failed)
        saved = store.get(job["id"])
        assert saved["push_status"] == "pushed"
        assert saved["push_payload"] == before
        assert json.loads(saved["receipt_json"])["data"]["taskId"] == job["id"]
        assert len(calls) == 1 and storage.count() == 1
        assert storage.rows()[0]["payload"] == json.loads(before)
    finally:
        await publisher.close()


@pytest.mark.asyncio
async def test_automatic_chat_delivery_only_enqueues_once_and_no_model_is_reexecuted(monkeypatch, tmp_path):
    monkeypatch.setenv("RESULT_PLATFORM_ENABLED", "1")
    monkeypatch.setenv("RESULT_PLATFORM_URL", "http://results.test/api/judgments")
    monkeypatch.setenv("RESULT_PLATFORM_TOKEN", LOCAL_TOKEN)
    monkeypatch.setenv("MAIN_FAN_AUTO_PUBLISH", "1")
    result = await service.analyze(complete_data(), "chat", "session")
    payload = result["platform_payload"]
    job = store.get(payload["taskId"])
    assert job["source"] == "chat" and job["push_status"] == "pending"
    assert job["status"] == "completed"
    assert store.claim("main_fan") is None
    store.record_completed_result("main_fan", "chat", "session", {}, result, "http://results.test/api/judgments")
    assert len(store.list_jobs("main_fan", "chat")) == 1
    publisher = ResultPlatformClient(settings(), transport=httpx.ASGITransport(app=create_app(ResultStore(tmp_path / "received.sqlite"))))
    try:
        await BusinessTaskController(publisher=publisher).publish_job(job)
        assert store.get(job["id"])["push_status"] == "pushed"
    finally:
        await publisher.close()


@pytest.mark.asyncio
async def test_changed_destination_blocks_old_pending_payload(tmp_path):
    result = await service.analyze(complete_data(), "test", "session")
    payload = result["platform_payload"]
    job = store.record_completed_result("main_fan", "test", "session", {}, result, "http://old.test/api/judgments")
    def forbidden(request):
        raise AssertionError("配置变化后不能自动向新地址发送历史结果")
    publisher = ResultPlatformClient(settings(), transport=httpx.MockTransport(forbidden))
    try:
        await BusinessTaskController(publisher=publisher).publish_job(job)
        assert store.get(job["id"])["push_status"] == "failed"
        assert json.loads(store.get(job["id"])["push_payload"])["taskId"] == payload["taskId"]
    finally:
        await publisher.close()


def test_publish_configuration_is_independent_from_data_gateway(monkeypatch):
    monkeypatch.setenv("RESULT_PLATFORM_ENABLED", "1")
    monkeypatch.setenv("SHANXI_GATEWAY_ENABLED", "0")
    job = submit("main_fan", complete_data(), publish=True)
    assert job["status"] == "queued"
    monkeypatch.setenv("RESULT_PLATFORM_ENABLED", "0")
    with pytest.raises(ValueError, match="结果接收平台"):
        submit("main_fan", complete_data(), publish=True)


@pytest.mark.asyncio
async def test_client_checks_business_receipts_and_rate_limit_without_gateway_credentials():
    payload = await result_payload()
    replies = [httpx.Response(429, headers={"Retry-After": "120"}),
               httpx.Response(200, json={"code": 500, "success": False}),
               httpx.Response(200, text="login page"), httpx.Response(302, headers={"Location": "http://elsewhere.test"}),
               httpx.Response(200, json={"code": 200, "data": {"taskId": "other"}}), httpx.Response(204)]
    def handler(request):
        assert request.headers["Authorization"] == f"Bearer {LOCAL_TOKEN}"
        assert request.url.host == "results.test"
        assert "X-Agent-Code" not in request.headers
        return replies.pop(0)
    client = ResultPlatformClient(settings(), transport=httpx.MockTransport(handler))
    try:
        with pytest.raises(ResultPlatformError) as limited:
            await client.push(payload)
        assert limited.value.retryable and limited.value.retry_after == 120
        for _ in range(4):
            with pytest.raises(ResultPlatformError):
                await client.push(payload)
        assert (await client.push(payload))["httpStatus"] == 204
    finally:
        await client.close()


@pytest.mark.asyncio
async def test_custom_auth_header_and_success_contract():
    payload = await result_payload()
    def handler(request):
        assert request.headers["X-API-Key"] == "result-secret"
        assert "Authorization" not in request.headers
        return httpx.Response(200, json={"status": {"value": "OK"}})
    client = ResultPlatformClient(ResultPlatformSettings(url="http://results.test/save", token="result-secret",
        auth_header="X-API-Key", auth_scheme="", success_field="status.value", success_values=("OK",)),
        transport=httpx.MockTransport(handler))
    try:
        assert (await client.push(payload))["status"]["value"] == "OK"
    finally:
        await client.close()


@pytest.mark.asyncio
async def test_export_and_publish_api_respect_saved_result_and_owner(monkeypatch, tmp_path):
    from main import app
    from runtime import business_tasks
    monkeypatch.setenv("BUSINESS_API_TOKEN", "business-secret")
    result = await service.analyze(complete_data(), "owner", "session")
    job = store.record_completed_result("main_fan", "owner", "session", {}, result, "http://results.test/api/judgments")
    publisher = ResultPlatformClient(settings(), transport=httpx.ASGITransport(app=create_app(ResultStore(tmp_path / "received.sqlite"))))
    monkeypatch.setattr(business_tasks, "_controller", BusinessTaskController(publisher=publisher))
    base = f"/api/business-agents/main_fan/jobs/{job['id']}"
    try:
        async with httpx.AsyncClient(transport=httpx.ASGITransport(app=app), base_url="http://agent.test") as http:
            headers = {"Authorization": "Bearer business-secret"}
            assert (await http.get(base + "/result")).status_code == 401
            assert (await http.get(base + "/result", headers=headers, params={"user_name": "other"})).status_code == 404
            exported = await http.get(base + "/result", headers=headers, params={"user_name": "owner"})
            assert exported.json() == result["platform_payload"]
            again = await http.post(base + "/publish", headers=headers, params={"user_name": "owner"})
            assert again.status_code == 202 and again.json()["push_status"] == "pending"
            analysis = await http.get(f"/api/business-agents/main_fan/analyses/{result['analysis_id']}/result", headers=headers,
                                      params={"session_id": "session", "user_name": "owner"})
            assert analysis.json() == result["platform_payload"]
    finally:
        await publisher.close()