"""结果平台独立性、业务编码、固定报文重试和 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()