import json import logging from types import SimpleNamespace from fastapi import FastAPI from fastapi.testclient import TestClient from app.gateway.routers import html_runtime from deerflow.agents.lead_agent import agent as agent_module from deerflow.agents.lead_agent import prompt as prompt_module from deerflow.config.paths import Paths from deerflow.config.skills_config import EsQueryRoutingConfig, SkillsConfig from deerflow.tools.builtins import page_report_template_tool as page_tool from deerflow.tools.builtins.page_report_template_tool import ( _brief_layout_mode, _brief_page_count, _brief_visualizations, _choose_page_profile, _quality_report, _requested_layout_mode, _requested_page_count, _required_visualizations, ) def _app_config(*, enabled: bool, prompt: str = "自定义 ES 路由规则") -> SimpleNamespace: return SimpleNamespace( skills=SkillsConfig( es_query_routing=EsQueryRoutingConfig(enabled=enabled, prompt=prompt), ), html_runtime=SimpleNamespace( base_url="http://localhost:8001/api/public/html-runtime" ), ) def test_es_query_routing_config_is_disabled_by_default() -> None: config = SkillsConfig() assert config.es_query_routing.enabled is False assert "es_query" in config.es_query_routing.prompt def test_es_query_routing_section_honors_switch_prompt_and_allowlist() -> None: enabled = _app_config(enabled=True, prompt=" 自定义 ES 路由规则 ") assert prompt_module._build_es_query_routing_section(app_config=enabled) == ( '\n自定义 ES 路由规则\n' ) assert ( prompt_module._build_es_query_routing_section( app_config=enabled, available_skills={"other_skill"}, ) == "" ) assert "自定义 ES 路由规则" in prompt_module._build_es_query_routing_section( app_config=enabled, available_skills={"es_query"}, ) assert prompt_module._build_es_query_routing_section(app_config=_app_config(enabled=False)) == "" assert prompt_module._build_es_query_routing_section(app_config=_app_config(enabled=True, prompt=" ")) == "" def test_es_query_routing_suppresses_explicitly_offline_skill(tmp_path, monkeypatch) -> None: extensions_path = tmp_path / "extensions_config.json" extensions_path.write_text( json.dumps({"mcpServers": {}, "skills": {"es_query": {"enabled": False}}}), encoding="utf-8", ) monkeypatch.setenv("DEER_FLOW_EXTENSIONS_CONFIG_PATH", str(extensions_path)) assert prompt_module._build_es_query_routing_section(app_config=_app_config(enabled=True)) == "" extensions_path.write_text( json.dumps({"mcpServers": {}, "skills": {"es_query": {"enabled": True}}}), encoding="utf-8", ) assert '' in prompt_module._build_es_query_routing_section( app_config=_app_config(enabled=True), ) def test_es_query_routing_is_injected_into_lead_and_allowed_custom_agent_prompts(monkeypatch) -> None: config = _app_config(enabled=True) monkeypatch.setattr(prompt_module, "_get_memory_context", lambda *args, **kwargs: "") monkeypatch.setattr(prompt_module, "get_agent_soul", lambda *args, **kwargs: "") monkeypatch.setattr(prompt_module, "get_skills_prompt_section", lambda *args, **kwargs: "") monkeypatch.setattr(prompt_module, "get_deferred_tools_prompt_section", lambda *args, **kwargs: "") monkeypatch.setattr(prompt_module, "_build_acp_section", lambda *args, **kwargs: "") monkeypatch.setattr(prompt_module, "_build_custom_mounts_section", lambda *args, **kwargs: "") monkeypatch.setattr(prompt_module, "_current_time_context", lambda: "") lead_prompt = prompt_module.apply_prompt_template(app_config=config) allowed_custom_prompt = prompt_module.apply_agent_only_prompt_template( agent_id="agent-with-es", agent_name="Agent with ES", available_skills={"es_query"}, app_config=config, ) denied_custom_prompt = prompt_module.apply_agent_only_prompt_template( agent_id="agent-without-es", agent_name="Agent without ES", available_skills={"other_skill"}, app_config=config, ) assert '' in lead_prompt assert "自定义 ES 路由规则" in lead_prompt assert '' in allowed_custom_prompt assert '' not in denied_custom_prompt def test_es_query_routing_registration_log_confirms_exact_final_prompt(caplog) -> None: config = _app_config(enabled=True, prompt="自定义 ES 路由规则") system_prompt = '\n自定义 ES 路由规则\n' with caplog.at_level(logging.INFO, logger=agent_module.__name__): agent_module._log_es_query_routing_registration( system_prompt=system_prompt, app_config=config, agent_id=None, available_skills=None, ) assert "ES query routing prompt registration" in caplog.text assert "agent_id=default" in caplog.text assert "enabled=True" in caplog.text assert "eligible=True" in caplog.text assert "registered=True" in caplog.text assert "prompt_chars=11" in caplog.text assert "prompt_sha256=" in caplog.text def test_es_query_routing_registration_log_explains_custom_agent_exclusion(caplog) -> None: config = _app_config(enabled=True, prompt="自定义 ES 路由规则") with caplog.at_level(logging.INFO, logger=agent_module.__name__): agent_module._log_es_query_routing_registration( system_prompt="custom agent prompt", app_config=config, agent_id="agent-without-es", available_skills={"other_skill"}, ) assert "agent_id=agent-without-es" in caplog.text assert "enabled=True" in caplog.text assert "eligible=False" in caplog.text assert "registered=False" in caplog.text def test_formal_markdown_rules_are_scoped_to_ordinary_qa(monkeypatch) -> None: config = _app_config(enabled=False) monkeypatch.setattr(prompt_module, "_ordinary_qa_markdown_format_enabled", lambda: True) monkeypatch.setattr(prompt_module, "_get_memory_context", lambda *args, **kwargs: "") monkeypatch.setattr(prompt_module, "get_agent_soul", lambda *args, **kwargs: "") monkeypatch.setattr(prompt_module, "get_skills_prompt_section", lambda *args, **kwargs: "") monkeypatch.setattr(prompt_module, "get_deferred_tools_prompt_section", lambda *args, **kwargs: "") monkeypatch.setattr(prompt_module, "_build_acp_section", lambda *args, **kwargs: "") monkeypatch.setattr(prompt_module, "_build_custom_mounts_section", lambda *args, **kwargs: "") monkeypatch.setattr(prompt_module, "_current_time_context", lambda: "") ordinary_prompt = prompt_module.apply_prompt_template(app_config=config) writing_prompt = prompt_module.apply_prompt_template(app_config=config, writing_mode=True) notebook_prompt = prompt_module.apply_prompt_template(app_config=config, is_notebook_mode=True) scheduled_prompt = prompt_module.apply_prompt_template(app_config=config, is_scheduled_run=True) assert '' in ordinary_prompt assert "## 一、标题" in ordinary_prompt assert "### (一) 标题" in ordinary_prompt assert "#### 1. 标题" in ordinary_prompt assert "2.1.1" in ordinary_prompt assert "无序圆点列表" in ordinary_prompt assert '' not in writing_prompt assert '' not in notebook_prompt assert '' not in scheduled_prompt def test_formal_markdown_rules_are_disabled_when_switch_is_off(monkeypatch) -> None: config = _app_config(enabled=False) monkeypatch.setattr(prompt_module, "_ordinary_qa_markdown_format_enabled", lambda: False) monkeypatch.setattr(prompt_module, "_get_memory_context", lambda *args, **kwargs: "") monkeypatch.setattr(prompt_module, "get_agent_soul", lambda *args, **kwargs: "") monkeypatch.setattr(prompt_module, "get_skills_prompt_section", lambda *args, **kwargs: "") monkeypatch.setattr(prompt_module, "get_deferred_tools_prompt_section", lambda *args, **kwargs: "") monkeypatch.setattr(prompt_module, "_build_acp_section", lambda *args, **kwargs: "") monkeypatch.setattr(prompt_module, "_build_custom_mounts_section", lambda *args, **kwargs: "") monkeypatch.setattr(prompt_module, "_current_time_context", lambda: "") ordinary_prompt = prompt_module.apply_prompt_template(app_config=config) assert '' not in ordinary_prompt def test_message_visualization_rules_are_offline_safe_and_scoped_to_ordinary_qa(monkeypatch) -> None: config = _app_config(enabled=False) monkeypatch.setattr(prompt_module, "_get_memory_context", lambda *args, **kwargs: "") monkeypatch.setattr(prompt_module, "get_agent_soul", lambda *args, **kwargs: "") monkeypatch.setattr(prompt_module, "get_skills_prompt_section", lambda *args, **kwargs: "") monkeypatch.setattr(prompt_module, "get_deferred_tools_prompt_section", lambda *args, **kwargs: "") monkeypatch.setattr(prompt_module, "_build_acp_section", lambda *args, **kwargs: "") monkeypatch.setattr(prompt_module, "_build_custom_mounts_section", lambda *args, **kwargs: "") monkeypatch.setattr(prompt_module, "_current_time_context", lambda: "") ordinary_prompt = prompt_module.apply_prompt_template(app_config=config) writing_prompt = prompt_module.apply_prompt_template(app_config=config, writing_mode=True) notebook_prompt = prompt_module.apply_prompt_template(app_config=config, is_notebook_mode=True) scheduled_prompt = prompt_module.apply_prompt_template(app_config=config, is_scheduled_run=True) assert "仅当用户明确要求在普通问答中展示地图" in ordinary_prompt assert "map-json" in ordinary_prompt assert "不得为了展示地图创建或写入 HTML 文件" in ordinary_prompt assert "用户只要求画流程图时,必须直接在当前消息中输出 Mermaid" in ordinary_prompt assert "用户只要求画图表时,必须沿用原有的消息内图表回答方式" in ordinary_prompt assert "analysis-mindmap" in ordinary_prompt assert "mindmap-json" in ordinary_prompt assert "普通分析、总结、提纲、结构化 JSON、流程图或图表都不得自行升级为思维导图" in ordinary_prompt assert "禁止读取或调用 `page-report-generation`" in ordinary_prompt assert "默认使用消息内展示" in ordinary_prompt assert "任务就必须作为文件交付" in ordinary_prompt assert "调用 `present_files` 打开制品预览" in ordinary_prompt assert "禁止在消息正文中直接输出完整 HTML 源码" in ordinary_prompt assert "HTML 页面默认必须是单页面" in ordinary_prompt assert "报告章节很多、已有 Markdown 很长,都不能作为生成多页面的理由" in ordinary_prompt assert "flow-json" not in ordinary_prompt assert "chart-json" not in ordinary_prompt assert "禁止在地图数据中放入 URL" in ordinary_prompt assert "map-json" not in writing_prompt assert "map-json" not in notebook_prompt assert "map-json" not in scheduled_prompt assert "mindmap-json" not in writing_prompt assert "mindmap-json" not in notebook_prompt assert "mindmap-json" not in scheduled_prompt def test_html_artifact_visualizations_use_local_declarative_contracts(monkeypatch) -> None: config = _app_config(enabled=False) monkeypatch.setattr(prompt_module, "_get_memory_context", lambda *args, **kwargs: "") monkeypatch.setattr(prompt_module, "get_agent_soul", lambda *args, **kwargs: "") monkeypatch.setattr(prompt_module, "get_skills_prompt_section", lambda *args, **kwargs: "") monkeypatch.setattr(prompt_module, "get_deferred_tools_prompt_section", lambda *args, **kwargs: "") monkeypatch.setattr(prompt_module, "_build_acp_section", lambda *args, **kwargs: "") monkeypatch.setattr(prompt_module, "_build_custom_mounts_section", lambda *args, **kwargs: "") monkeypatch.setattr(prompt_module, "_current_time_context", lambda: "") ordinary_prompt = prompt_module.apply_prompt_template(app_config=config) writing_prompt = prompt_module.apply_prompt_template(app_config=config, writing_mode=True) assert '' in ordinary_prompt assert "只有通过 `` 中的 HTML 页面意图门禁后" in ordinary_prompt assert "不得只在聊天消息中粘贴 HTML 代码" in ordinary_prompt assert "不得仅因为用户要求画地图" in ordinary_prompt assert "普通地图请求必须在消息列表中使用 `map-json`" in ordinary_prompt assert 'data-df-flow="process-flow"' in ordinary_prompt assert "流程图不要使用 Mermaid" in ordinary_prompt assert "AntV G6" in ordinary_prompt assert 'data-df-chart="sales-chart"' in ordinary_prompt assert 'data-df-geo="area-map"' in ordinary_prompt assert '/api/public/html-runtime/g6.min.js' in ordinary_prompt assert '/api/public/html-runtime/echarts.min.js' in ordinary_prompt assert '/api/public/html-runtime/deerflow-runtime.js' in ordinary_prompt assert "不要在 HTML 中省略这些引用" in ordinary_prompt assert '"map":"china-provinces"' in ordinary_prompt assert "禁止 Leaflet、`data-df-map`" in ordinary_prompt assert "地图瓦片 URL" in ordinary_prompt assert "deerflow:geo-map-click" in ordinary_prompt assert "us-states" in ordinary_prompt assert "japan-prefectures" in ordinary_prompt assert "taiwan-counties" in ordinary_prompt assert "详情浮层" in ordinary_prompt assert "pointer-events:none" in ordinary_prompt assert "普通问答消息列表的 `map-json` 保持原有前端瓦片渲染" in ordinary_prompt assert "单页面”表示一个可以自然纵向滚动的 HTML 文档" in ordinary_prompt assert "禁止固定 1920×1080 舞台" in ordinary_prompt assert "默认禁止读取 `page-template-library`" in ordinary_prompt assert "第一项文件操作必须是用 `write_file` 创建" in ordinary_prompt assert "默认直接用 `write_file` 在 `/mnt/user-data/outputs/`" in ordinary_prompt assert "data-df-chart` 的内容必须能被 `JSON.parse` 直接解析" in ordinary_prompt assert "中间地图舞台" in ordinary_prompt assert "中间至少 65% 的地图画面保持无遮挡" in ordinary_prompt assert '' not in writing_prompt def test_html_runtime_exposes_only_local_geojson_maps(monkeypatch) -> None: monkeypatch.setattr( html_runtime, "get_app_config", lambda: SimpleNamespace( html_runtime=SimpleNamespace( base_url="http://gateway.internal/api/public/html-runtime" ) ), ) app = FastAPI() app.include_router(html_runtime.router) client = TestClient(app) runtime = client.get("/api/public/html-runtime/deerflow-runtime.js") assert runtime.status_code == 200 assert "data-df-geo" in runtime.text assert "df-geo-popup" in runtime.text assert "deerflow:geo-map-click" in runtime.text assert "mapTileUrl" not in runtime.text assert client.get("/api/public/html-runtime/leaflet.js").status_code == 404 for map_id, filename, minimum_features in ( ("china-provinces", "china-provinces.geojson", 30), ("us-states", "us-states.geojson", 51), ("japan-prefectures", "japan-prefectures.geojson", 40), ("taiwan-counties", "taiwan-counties.geojson", 20), ): assert f"/maps/{filename}" in runtime.text geojson = client.get(f"/api/public/html-runtime/maps/{filename}") assert geojson.status_code == 200 assert "application/geo+json" in geojson.headers["content-type"] assert geojson.json()["type"] == "FeatureCollection" assert len(geojson.json()["features"]) >= minimum_features def _report_html(body: str, scripts: str = "") -> str: return ( "测试报告" f"{scripts}

{'报告正文' * 180}

{body}" ) def test_page_brief_preserves_markdown_visualization_requirements() -> None: assert _required_visualizations( [], "Markdown 报告包含 Mermaid 流程图、销售额折线图和城市坐标点地图。", ) == {"flowchart", "chart", "map"} def test_page_quality_rejects_manual_flow_and_accepts_g6_contract() -> None: manual = _quality_report( _report_html('

发展流程图

技术突破
'), data_mode="provided", expected_features={"flowchart"}, ) compliant = _quality_report( _report_html( '
' '', '', ), data_mode="provided", expected_features={"flowchart"}, ) assert manual["ok"] is False assert any("流程图未使用平台绘图契约" in error for error in manual["errors"]) assert compliant["ok"] is True def test_page_quality_checks_chart_and_map_runtime_contracts() -> None: report = _quality_report( _report_html("

销售图表与城市地图

"), data_mode="provided", expected_features={"chart", "map"}, ) assert report["ok"] is False assert any("图表未使用平台绘图契约" in error for error in report["errors"]) assert any("地图未使用平台绘图契约" in error for error in report["errors"]) def test_page_quality_rejects_non_json_echarts_callbacks() -> None: scripts = '' report = _quality_report( _report_html( '
' '', scripts, ), data_mode="provided", expected_features={"chart"}, ) assert report["ok"] is False assert any("data-df-chart 不是严格 JSON" in error for error in report["errors"]) def test_page_quality_accepts_strict_json_echarts_options() -> None: scripts = '' payload = json.dumps( { "tooltip": {"formatter": "{b}: {c}"}, "series": [ { "type": "bar", "data": [{"value": 12, "itemStyle": {"color": "#0ea5e9"}}], } ], } ) report = _quality_report( _report_html( '
' f'', scripts, ), data_mode="provided", expected_features={"chart"}, ) assert report["ok"] is True def test_page_quality_accepts_each_catalog_geojson_map() -> None: scripts = '' for map_name in ("china-provinces", "us-states", "japan-prefectures", "taiwan-counties"): report = _quality_report( _report_html( '
' '", scripts, ), data_mode="provided", expected_features={"map"}, ) assert report["ok"] is True def test_page_validation_restores_prepare_brief_requirements(tmp_path) -> None: actual = tmp_path / "人工智能发展报告.html" actual.write_text(_report_html("

报告页面

"), encoding="utf-8") brief = { "requested_page_count": 1, "required_features": [], "required_visualizations": ["flowchart"], "user_material_brief": "原 Markdown 中包含 Mermaid 流程图。", } (tmp_path / "人工智能发展报告--page-brief.json").write_text( json.dumps(brief, ensure_ascii=False), encoding="utf-8", ) assert _brief_visualizations(actual, tmp_path) == {"flowchart"} assert _brief_page_count(actual, tmp_path) == 1 def test_page_template_defaults_to_single_page_and_overrides_multi_template() -> None: requested_page_count = _requested_page_count(None) profile = _choose_page_profile( theme="dark", content_density="text-heavy", required_features=["text-panels"], requested_page_count=requested_page_count, selection_key="default-single-page", exclude_template_ids=[], template_id="dark-charts-portal", ) assert requested_page_count == 1 assert profile.page_count == 1 assert profile.id == "dark-document" def test_page_template_allows_multi_page_only_with_explicit_count() -> None: profile = _choose_page_profile( theme="light", content_density="text-heavy", required_features=["text-panels"], requested_page_count=3, selection_key="explicit-multi-page", exclude_template_ids=[], template_id="light-airy", ) assert profile.page_count > 1 def test_page_quality_rejects_multi_view_html_when_single_page_is_required() -> None: report = _quality_report( _report_html( '' '
第一页
第二页
' ), data_mode="provided", expected_features=set(), expected_page_count=1, ) assert report["ok"] is False assert any("HTML 必须保持单页面" in error for error in report["errors"]) def test_page_layout_defaults_to_document_unless_large_screen_is_explicit() -> None: assert _requested_layout_mode("把这份 Markdown 画成一个 HTML 页面") == "document" assert _requested_layout_mode("制作 1920×1080 单屏驾驶舱大屏") == "screen" def test_document_quality_rejects_fixed_canvas_and_undersized_long_flow() -> None: nodes = [{"id": str(index), "label": f"节点{index}"} for index in range(9)] payload = json.dumps({"direction": "LR", "nodes": nodes, "edges": []}, ensure_ascii=False) report = _quality_report( _report_html( '' '
' f'', '', ), data_mode="provided", expected_features={"flowchart"}, expected_layout_mode="document", ) assert report["ok"] is False assert any("overflow:hidden" in error for error in report["errors"]) assert any("1920×1080" in error for error in report["errors"]) assert any("至少需要 360px" in error for error in report["errors"]) assert any("改用 TB" in error for error in report["errors"]) def test_prepare_places_live_editable_html_in_outputs(tmp_path, monkeypatch) -> None: paths = Paths(tmp_path / ".deer-flow") monkeypatch.setattr(page_tool, "get_paths", lambda: paths) monkeypatch.setattr(page_tool, "resolve_path_user_id", lambda _thread_id: "tester") runtime = SimpleNamespace( context={"thread_id": "page-preview-test"}, config={"configurable": {"thread_id": "page-preview-test"}}, ) prepared = json.loads( page_tool._prepare( runtime, theme="light", content_density="text-heavy", required_features=["flowchart"], preferred_page_count=1, selection_key="preview", exclude_template_ids=[], template_id="light-document", output_name="人工智能发展报告.html", page_brief="把 Markdown 报告生成普通单页 HTML,包含流程图。", ) ) assert prepared["working_copy"] == "/mnt/user-data/outputs/人工智能发展报告.html" assert prepared["working_copy"] == prepared["final_output"] actual = paths.resolve_virtual_path( "page-preview-test", prepared["working_copy"], user_id="tester" ) brief_path = paths.resolve_virtual_path( "page-preview-test", prepared["brief_path"], user_id="tester" ) assert actual.is_file() assert _brief_layout_mode(actual, brief_path.parent) == "document"