598 lines
27 KiB
Python
598 lines
27 KiB
Python
"""结构化 HTML 模板(kind=structured,先拿 no.1 落地)的引擎/存储/路由测试。
|
|
|
|
覆盖:
|
|
* seed/meta:结构化 no1 四件套正确入库,单文件模板(如 no2)不受影响;
|
|
* 契约语义与单文件时代等价:``reference_data`` 读 data.json 与旧 ``const DATA=`` 行
|
|
解析结果一致,``validate_data`` / ``merge_with_reference`` 行为不变;
|
|
* 组合:``read_template_text`` / ``fill_template`` 产出无残留标记的自包含文档,
|
|
数据注入可被页面 ``JSON.parse`` 还原,标题/副标题/更新时间生效;
|
|
* data.json 持久化(管理页「编辑数据」):整体替换 + 契约校验(顶层键不可增删);
|
|
* 骨架编辑标记校验、结构化重置;
|
|
* 旧单文件存储副本的自动升级:未手改 → 升级;手改过 → 保留旧版;
|
|
* 路由:content 的 doc/raw 双形态、data 的 GET/PUT 与权限。
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import json
|
|
import re
|
|
from types import SimpleNamespace
|
|
from typing import Any
|
|
|
|
import pytest
|
|
from fastapi import HTTPException
|
|
from test_api_read_cache import FakeRedis, _config
|
|
|
|
from app.gateway import api_read_cache
|
|
from app.gateway.routers import html_templates
|
|
from deerflow.runtime.scheduler import template_fill, template_store
|
|
|
|
# no1 是当前唯一的结构化模板。
|
|
TPL = "no1"
|
|
LEGACY_FILE = template_fill._TEMPLATES_ROOT / TPL / "taiwan_littoral_ui_redwhite_v2.html"
|
|
|
|
|
|
@pytest.fixture
|
|
def tmp_store(monkeypatch, tmp_path):
|
|
"""隔离的运行时存储:全部用例不触碰真实 .deer-flow 目录。"""
|
|
root = tmp_path / "html-templates"
|
|
monkeypatch.setattr(template_store, "store_root", lambda: root)
|
|
template_fill.invalidate_cache()
|
|
template_store.ensure_seeded()
|
|
yield root
|
|
template_fill.invalidate_cache()
|
|
|
|
|
|
@pytest.fixture
|
|
def api_request(monkeypatch, tmp_store):
|
|
"""可直接调路由函数的最小 request(admin 身份,目录缓存走 FakeRedis)。"""
|
|
redis = FakeRedis()
|
|
monkeypatch.setattr("deerflow.persistence.engine.get_session_factory", lambda: None)
|
|
monkeypatch.setattr(api_read_cache, "_api_read_cache_settings_cache", None)
|
|
monkeypatch.setattr(api_read_cache, "_client", lambda *a, **k: (redis, "test:catalog"))
|
|
request = SimpleNamespace(
|
|
app=SimpleNamespace(state=SimpleNamespace(config=_config())),
|
|
state=SimpleNamespace(user=SimpleNamespace(id="admin-1", system_role="admin")),
|
|
)
|
|
|
|
async def current_user(req):
|
|
return req.state.user
|
|
|
|
monkeypatch.setattr(html_templates, "get_optional_user_from_request", current_user)
|
|
return request
|
|
|
|
|
|
def _page_data_element(html: str) -> dict:
|
|
match = re.search(
|
|
r'<script type="application/json" id="page-data">(.*?)</script>', html, re.DOTALL
|
|
)
|
|
assert match is not None, "组合产物缺少 page-data JSON 元素"
|
|
return json.loads(match.group(1))
|
|
|
|
|
|
# ── seed / meta ──────────────────────────────────────────────────────────────
|
|
|
|
|
|
def test_structured_seed_and_single_untouched(tmp_store):
|
|
meta = template_store.get_template_meta(TPL)
|
|
assert meta is not None
|
|
assert meta["kind"] == "structured"
|
|
assert meta["convertible"] is True
|
|
assert meta["manifest_var"] == "MANIFEST"
|
|
assert sorted(p.name for p in (tmp_store / TPL).iterdir()) == [
|
|
"data.json",
|
|
"fields.json",
|
|
"manifest.json",
|
|
"meta.json",
|
|
"runtime.js",
|
|
"template.html",
|
|
]
|
|
|
|
# 单文件模板照旧(kind=single,正文 = 完整文档,无结构化部件)。
|
|
no2 = template_store.get_template_meta("no2")
|
|
assert no2 is not None and no2["kind"] == "single"
|
|
assert not (tmp_store / "no2" / "runtime.js").exists()
|
|
|
|
|
|
def test_reference_data_matches_legacy_const_line(tmp_store):
|
|
"""data.json 的契约语义必须与单文件时代的 const DATA= 行完全一致。"""
|
|
legacy = template_fill._extract_const_object(LEGACY_FILE.read_text(encoding="utf-8"), "DATA")
|
|
ref = template_fill.reference_data(TPL)
|
|
assert ref == legacy
|
|
assert template_fill.validate_data(TPL, legacy) == []
|
|
assert set(template_fill.top_level_keys(TPL)) == set(legacy.keys())
|
|
|
|
|
|
# ── 组合 / 填充 ─────────────────────────────────────────────────────────────
|
|
|
|
|
|
def test_read_template_text_composes_default_document(tmp_store):
|
|
html = template_fill.read_template_text(TPL)
|
|
for marker in template_fill.STRUCTURED_MARKERS:
|
|
assert marker not in html
|
|
assert 'id="page-data"' in html and 'id="page-manifest"' in html
|
|
assert "const DATA=JSON.parse" in html
|
|
# 页面内 JSON 编辑器随 runtime 内联。
|
|
assert 'id="editorOverlay"' in html and 'id="applyJson"' in html
|
|
assert _page_data_element(html)["overview"] == template_fill.reference_data(TPL)["overview"]
|
|
|
|
|
|
def test_fill_template_structured_injects_data_titles_and_date(tmp_store):
|
|
ref = template_fill.reference_data(TPL)
|
|
modified = json.loads(json.dumps(ref))
|
|
modified["overview"] = {
|
|
"核心指标": [{"名称": "测试指标", "数值": 7, "单位": "个", "说明": "单元测试"}]
|
|
}
|
|
import datetime as dt
|
|
|
|
html = template_fill.fill_template(
|
|
TPL,
|
|
modified,
|
|
update_date=dt.date(2026, 11, 4),
|
|
title="新标题",
|
|
subtitle="新副标题",
|
|
)
|
|
for marker in template_fill.STRUCTURED_MARKERS:
|
|
assert marker not in html
|
|
assert "<title>新标题</title>" in html
|
|
assert "<h1>新标题</h1>" in html
|
|
assert "新副标题" in html
|
|
# MANIFEST 系统.更新时间/名称 同步。
|
|
assert "2026-11-04" in html
|
|
assert '"名称": "新标题"' in html
|
|
# 注入的数据可被页面 JSON.parse 原样还原。
|
|
assert _page_data_element(html)["overview"]["核心指标"][0]["名称"] == "测试指标"
|
|
|
|
|
|
def test_merge_and_validate_semantics_unchanged(tmp_store):
|
|
# 顶层键缺失 → 硬校验报错(与单文件时代一致)。
|
|
errors = template_fill.validate_data(TPL, {"overview": {}})
|
|
assert any("缺少顶层字段" in e for e in errors)
|
|
# 空内容分支回退模板默认数据。
|
|
merged = template_fill.merge_with_reference(TPL, {"overview": {}, "regions": []})
|
|
assert merged == template_fill.reference_data(TPL)
|
|
# 有内容的分支保留转换结果。
|
|
merged2 = template_fill.merge_with_reference(
|
|
TPL, {"overview": {"核心指标": [{"名称": "x", "数值": 1, "单位": "", "说明": ""}]}}
|
|
)
|
|
assert merged2["overview"]["核心指标"][0]["名称"] == "x"
|
|
assert merged2["regions"] == template_fill.reference_data(TPL)["regions"]
|
|
|
|
|
|
# ── data.json 持久化 / 骨架编辑 / 重置 ──────────────────────────────────────
|
|
|
|
|
|
def test_write_read_template_data_roundtrip(tmp_store):
|
|
ref = template_fill.reference_data(TPL)
|
|
modified = json.loads(json.dumps(ref))
|
|
modified["overview"]["核心指标"][0]["名称"] = "持久化指标"
|
|
meta = template_store.write_template_data(TPL, modified)
|
|
assert meta["kind"] == "structured"
|
|
assert template_store.read_template_data(TPL)["overview"]["核心指标"][0]["名称"] == "持久化指标"
|
|
# 引擎缓存已失效:reference_data / 组合产物立即反映新数据。
|
|
assert template_fill.reference_data(TPL)["overview"]["核心指标"][0]["名称"] == "持久化指标"
|
|
assert "持久化指标" in template_fill.read_template_text(TPL)
|
|
|
|
|
|
def test_write_template_manifest_persists_and_composes(tmp_store):
|
|
"""MANIFEST 持久化(系统名称/导航页面/交互规则):写回 manifest.json 并进入组合产物。"""
|
|
ref = template_fill.reference_data(TPL)
|
|
manifest = template_store.read_template_manifest(TPL)
|
|
assert manifest.get("系统", {}).get("名称") == "台湾滨海防卫知识数据展示系统"
|
|
manifest["系统"]["名称"] = "改名后的系统"
|
|
manifest["页面"][0]["名称"] = "首页"
|
|
template_store.write_template_data(TPL, ref, manifest=manifest)
|
|
assert template_store.read_template_manifest(TPL)["系统"]["名称"] == "改名后的系统"
|
|
# 组合产物(MANIFEST 内嵌 JSON)反映新值;导航按钮由页面 JS 依据 MANIFEST 渲染。
|
|
composed = template_fill.read_template_text(TPL)
|
|
assert "改名后的系统" in composed
|
|
assert '"名称": "首页"' in composed
|
|
# 非对象 MANIFEST 拒绝。
|
|
with pytest.raises(template_store.TemplateStoreError, match="MANIFEST"):
|
|
template_store.write_template_data(TPL, ref, manifest=["nope"])
|
|
# 重置回随包原版。
|
|
template_store.reset_template(TPL)
|
|
assert template_store.read_template_manifest(TPL)["系统"]["名称"] == "台湾滨海防卫知识数据展示系统"
|
|
|
|
|
|
def test_write_template_data_rejects_contract_changes(tmp_store):
|
|
with pytest.raises(template_store.TemplateStoreError, match="契约"):
|
|
template_store.write_template_data(TPL, {"overview": {}})
|
|
with pytest.raises(template_store.TemplateStoreError):
|
|
template_store.write_template_data(TPL, ["not", "an", "object"])
|
|
# 单文件模板不支持数据编辑。
|
|
with pytest.raises(template_store.TemplateStoreError, match="结构化"):
|
|
template_store.write_template_data("no2", {"x": 1})
|
|
|
|
|
|
def test_update_template_skeleton_marker_guard(tmp_store):
|
|
with pytest.raises(template_store.TemplateStoreError, match="注入标记"):
|
|
template_store.update_template(TPL, html="<!doctype html><p>标记丢了</p>")
|
|
# 保留标记的骨架编辑可以通过,且不影响 kind/convertible。
|
|
skeleton = template_store.read_template_html(TPL)
|
|
edited = skeleton.replace("JSON 编辑器", "JSON 编辑器·编辑", 1)
|
|
meta = template_store.update_template(TPL, html=edited)
|
|
assert meta["kind"] == "structured" and meta["convertible"] is True
|
|
assert template_fill.read_template_text(TPL).count("JSON 编辑器·编辑") == 1
|
|
|
|
|
|
def test_reset_structured_restores_all_parts(tmp_store):
|
|
template_store.write_template_data(
|
|
TPL, _with_first_metric_name("持久化指标")
|
|
)
|
|
template_store.update_template(TPL, html=template_store.read_template_html(TPL) + "\n<!-- edited -->")
|
|
template_store.reset_template(TPL)
|
|
assert template_store.read_template_data(TPL)["overview"]["核心指标"][0]["名称"] != "持久化指标"
|
|
assert "<!-- edited -->" not in template_store.read_template_html(TPL)
|
|
assert template_fill.read_template_text(TPL) # 组合仍可用
|
|
|
|
|
|
def _with_first_metric_name(name: str) -> dict:
|
|
ref = template_fill.reference_data(TPL)
|
|
modified = json.loads(json.dumps(ref))
|
|
modified["overview"]["核心指标"][0]["名称"] = name
|
|
return modified
|
|
|
|
|
|
# ── 旧单文件存储副本的自动升级 ─────────────────────────────────────────────
|
|
|
|
|
|
def _make_legacy_copy(store_root, *, modified: bool) -> None:
|
|
import shutil
|
|
|
|
tpl_dir = store_root / TPL
|
|
if tpl_dir.exists():
|
|
shutil.rmtree(tpl_dir)
|
|
tpl_dir.mkdir(parents=True)
|
|
text = LEGACY_FILE.read_text(encoding="utf-8")
|
|
if modified:
|
|
text += "\n<!-- hand edit -->"
|
|
(tpl_dir / "template.html").write_text(text, encoding="utf-8")
|
|
(tpl_dir / "meta.json").write_text(
|
|
json.dumps({"id": TPL, "name": "no.1", "builtin": True, "convertible": True}),
|
|
encoding="utf-8",
|
|
)
|
|
|
|
|
|
def test_auto_upgrade_untouched_legacy_copy(tmp_store):
|
|
_make_legacy_copy(tmp_store, modified=False)
|
|
assert template_store.ensure_seeded() is True
|
|
meta = template_store.get_template_meta(TPL)
|
|
assert meta["kind"] == "structured"
|
|
assert (tmp_store / TPL / "runtime.js").is_file()
|
|
# 升级后的模板可正常组合。
|
|
assert "__PAGE_" not in template_fill.read_template_text(TPL)
|
|
|
|
|
|
def test_auto_upgrade_skips_hand_modified_copy(tmp_store):
|
|
_make_legacy_copy(tmp_store, modified=True)
|
|
template_store.ensure_seeded()
|
|
meta = template_store.get_template_meta(TPL)
|
|
assert meta["kind"] == "single"
|
|
assert "<!-- hand edit -->" in (tmp_store / TPL / "template.html").read_text(encoding="utf-8")
|
|
# 管理页「重置」可手动升级到结构化版。
|
|
template_store.reset_template(TPL)
|
|
assert template_store.get_template_meta(TPL)["kind"] == "structured"
|
|
assert (tmp_store / TPL / "runtime.js").is_file()
|
|
|
|
|
|
# ── 路由 ────────────────────────────────────────────────────────────────────
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_router_content_doc_and_raw(tmp_store, api_request):
|
|
doc = await html_templates.get_html_template_content(TPL, api_request)
|
|
assert "__PAGE_DATA__" not in doc["html"]
|
|
assert 'id="page-data"' in doc["html"]
|
|
raw = await html_templates.get_html_template_content(TPL, api_request, raw=True)
|
|
assert "__PAGE_DATA__" in raw["html"] and "__PAGE_RUNTIME__" in raw["html"]
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_router_data_get_put_and_guards(tmp_store, api_request):
|
|
got = await html_templates.get_html_template_data(TPL, api_request)
|
|
assert "overview" in got["data"]
|
|
# MANIFEST 随 GET 返回(管理页「系统设置」模块)。
|
|
assert got["manifest"]["系统"]["名称"] == "台湾滨海防卫知识数据展示系统"
|
|
# 模块中文名(按 MANIFEST.页面 解析,管理页「编辑数据」的模块导航用)。
|
|
labels = {m["key"]: m["label"] for m in got["modules"]}
|
|
assert labels["overview"] == "总览"
|
|
assert labels["regions"] == "沿岸区域"
|
|
assert labels["organization_charts"] == "组织与单位"
|
|
assert labels["events"] == "演习与项目"
|
|
assert set(labels) == set(got["data"].keys())
|
|
|
|
payload = got["data"]
|
|
payload["overview"]["核心指标"][0]["名称"] = "路由持久化"
|
|
manifest = json.loads(json.dumps(got["manifest"]))
|
|
manifest["系统"]["名称"] = "路由改名系统"
|
|
meta = await html_templates.update_html_template_data(
|
|
TPL,
|
|
html_templates.TemplateDataUpdateRequest(data=payload, manifest=manifest),
|
|
api_request,
|
|
)
|
|
assert meta["kind"] == "structured"
|
|
assert template_store.read_template_data(TPL)["overview"]["核心指标"][0]["名称"] == "路由持久化"
|
|
assert template_store.read_template_manifest(TPL)["系统"]["名称"] == "路由改名系统"
|
|
# 仅传 data(未动 MANIFEST)时不回写 manifest.json。
|
|
await html_templates.update_html_template_data(
|
|
TPL, html_templates.TemplateDataUpdateRequest(data=payload), api_request
|
|
)
|
|
assert template_store.read_template_manifest(TPL)["系统"]["名称"] == "路由改名系统"
|
|
|
|
# 契约被破坏(删顶层键) → 400。
|
|
broken = {k: v for k, v in payload.items() if k != "regions"}
|
|
with pytest.raises(HTTPException) as exc:
|
|
await html_templates.update_html_template_data(
|
|
TPL, html_templates.TemplateDataUpdateRequest(data=broken), api_request
|
|
)
|
|
assert exc.value.status_code == 400
|
|
|
|
# 单文件模板 → 400。
|
|
with pytest.raises(HTTPException) as exc2:
|
|
await html_templates.get_html_template_data("no2", api_request)
|
|
assert exc2.value.status_code == 400
|
|
|
|
# 非管理员不可写。
|
|
api_request.state.user = SimpleNamespace(id="u2", system_role="user")
|
|
with pytest.raises(HTTPException) as exc3:
|
|
await html_templates.update_html_template_data(
|
|
TPL, html_templates.TemplateDataUpdateRequest(data=payload), api_request
|
|
)
|
|
assert exc3.value.status_code == 403
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_scheduler_fill_path_yields_selfcontained_html(tmp_store):
|
|
"""调度器模板填充路径的最终产物自包含:单文件、无标记、数据可解析。"""
|
|
data_obj = template_fill.merge_with_reference(TPL, {"overview": _with_first_metric_name("调度指标")["overview"]})
|
|
html = template_fill.fill_template(TPL, data_obj, update_date=__import__("datetime").date(2026, 1, 1))
|
|
assert html.startswith("<!doctype html>") and "__PAGE_" not in html
|
|
assert "调度指标" in html
|
|
assert _page_data_element(html)["overview"]["核心指标"][0]["名称"] == "调度指标"
|
|
|
|
|
|
# ── 字段显示 / 自定义模块 ───────────────────────────────────────────────────
|
|
|
|
|
|
def test_field_catalog_and_visibility_roundtrip(tmp_store):
|
|
catalog = template_store.read_template_fields(TPL)
|
|
labels = {f["label"] for f in catalog}
|
|
assert "节点类别" in labels and "来源" in labels
|
|
assert all(f.get("description") for f in catalog)
|
|
|
|
# 隐藏两个字段 → manifest 记 {label: false},其余全部显示。
|
|
template_store.write_template_field_visibility(TPL, ["节点类别", "来源"])
|
|
manifest = template_store.read_template_manifest(TPL)
|
|
assert manifest["字段显示"] == {"节点类别": False, "来源": False}
|
|
# 组合产物带隐藏配置,runtime 的 F() 据此不渲染。
|
|
composed = template_fill.read_template_text(TPL)
|
|
assert '"字段显示"' in composed
|
|
|
|
# 目录外字段拒绝。
|
|
with pytest.raises(template_store.TemplateStoreError, match="未知字段"):
|
|
template_store.write_template_field_visibility(TPL, ["不存在的字段"])
|
|
|
|
# 置空 = 全部显示。
|
|
template_store.write_template_field_visibility(TPL, [])
|
|
assert template_store.read_template_manifest(TPL)["字段显示"] == {}
|
|
|
|
|
|
def test_custom_modules_write_and_sanitize(tmp_store):
|
|
raw = (
|
|
'<div class="panel"><h3 onclick="evil()">x</h3><script>alert(1)</script>'
|
|
'<a href="javascript:alert(2)">链接</a><p>正常内容</p></div>'
|
|
)
|
|
template_store.write_template_custom_modules(
|
|
TPL, [{"编号": "c1", "页面": "overview", "标题": "测试模块", "html": raw}]
|
|
)
|
|
saved = template_store.read_template_manifest(TPL)["自定义模块"]
|
|
assert len(saved) == 1
|
|
html = saved[0]["html"]
|
|
assert "正常内容" in html
|
|
assert "<script" not in html and "onclick" not in html and "javascript:" not in html
|
|
# 组合产物包含模块内容与渲染函数。
|
|
composed = template_fill.read_template_text(TPL)
|
|
assert "正常内容" in composed and "renderCustomModules" in composed
|
|
# 结构校验:缺页面/缺 html 拒绝。
|
|
with pytest.raises(template_store.TemplateStoreError, match="页面"):
|
|
template_store.write_template_custom_modules(TPL, [{"html": "<div>x</div>"}])
|
|
with pytest.raises(template_store.TemplateStoreError, match="html"):
|
|
template_store.write_template_custom_modules(TPL, [{"页面": "overview"}])
|
|
|
|
|
|
def test_field_extract_locks_and_restore(tmp_store):
|
|
"""字段抽取锁定:锁定字段的键在抽取结果合并后恢复模板现有值(按条目身份匹配)。"""
|
|
ref = template_fill.reference_data(TPL)
|
|
|
|
# 锁定「记录类别」(org_units 条目的 记录类别 键)与「日期」(events 条目的 日期显示 键)。
|
|
template_store.write_template_field_locks(TPL, ["记录类别", "日期"])
|
|
keys = template_store.locked_field_keys(TPL)
|
|
assert "记录类别" in keys and "日期显示" in keys
|
|
|
|
# 模拟抽取结果:改了第一个单位的记录类别、事件日期,并保留身份键(id/名称)。
|
|
extracted = json.loads(json.dumps(ref))
|
|
unit = extracted["org_units"][0]
|
|
old_cat = ref["org_units"][0]["记录类别"]
|
|
unit["记录类别"] = "被抽取改过的类别"
|
|
event = extracted["events"][0]
|
|
event["日期显示"] = "2026-01-01"
|
|
event["名称"] = ref["events"][0].get("名称") or event.get("名称") or event.get("name") or "同身份"
|
|
|
|
restored = template_store.restore_locked_fields(TPL, extracted)
|
|
# 锁定键恢复旧值。
|
|
assert restored["org_units"][0]["记录类别"] == old_cat
|
|
assert restored["events"][0]["日期显示"] == ref["events"][0]["日期显示"]
|
|
# 未锁定分支不动。
|
|
assert restored["overview"] == extracted["overview"]
|
|
# 目录外字段拒绝。
|
|
with pytest.raises(template_store.TemplateStoreError, match="未知字段"):
|
|
template_store.write_template_field_locks(TPL, ["不存在字段"])
|
|
# 清空 = 全部接受抽取。
|
|
template_store.write_template_field_locks(TPL, [])
|
|
assert template_store.restore_locked_fields(TPL, extracted) == extracted
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_router_fields_extract_flags(tmp_store, api_request):
|
|
got = await html_templates.get_html_template_fields(TPL, api_request)
|
|
by_label = {f["label"]: f for f in got["fields"]}
|
|
assert by_label["记录类别"]["extract"] is True
|
|
assert by_label["记录类别"]["has_key"] is True
|
|
assert by_label["组织与单位概况"]["has_key"] is False # 布局分区
|
|
|
|
meta = await html_templates.update_html_template_fields(
|
|
TPL,
|
|
html_templates.FieldVisibilityUpdateRequest(hidden=[], extract_locked=["记录类别"]),
|
|
api_request,
|
|
)
|
|
assert meta["kind"] == "structured"
|
|
got2 = await html_templates.get_html_template_fields(TPL, api_request)
|
|
by2 = {f["label"]: f for f in got2["fields"]}
|
|
assert by2["记录类别"]["extract"] is False and by2["日期"]["extract"] is True
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_router_fields_and_custom_modules(tmp_store, api_request):
|
|
got = await html_templates.get_html_template_fields(TPL, api_request)
|
|
assert any(f["label"] == "节点类别" and f["visible"] for f in got["fields"])
|
|
assert [p["key"] for p in got["pages"]][0] == "overview"
|
|
assert got["custom_modules"] == []
|
|
|
|
meta = await html_templates.update_html_template_fields(
|
|
TPL, html_templates.FieldVisibilityUpdateRequest(hidden=["节点类别"]), api_request
|
|
)
|
|
assert meta["kind"] == "structured"
|
|
got2 = await html_templates.get_html_template_fields(TPL, api_request)
|
|
by_label = {f["label"]: f["visible"] for f in got2["fields"]}
|
|
assert by_label["节点类别"] is False and by_label["来源"] is True
|
|
|
|
await html_templates.update_html_template_custom_modules(
|
|
TPL,
|
|
html_templates.CustomModuleListUpdateRequest(
|
|
modules=[{"页面": "overview", "标题": "t", "html": "<div>hi</div>"}]
|
|
),
|
|
api_request,
|
|
)
|
|
got3 = await html_templates.get_html_template_fields(TPL, api_request)
|
|
assert got3["custom_modules"][0]["html"] == "<div>hi</div>"
|
|
|
|
# 非管理员不可写。
|
|
api_request.state.user = SimpleNamespace(id="u2", system_role="user")
|
|
with pytest.raises(HTTPException) as exc:
|
|
await html_templates.update_html_template_fields(
|
|
TPL, html_templates.FieldVisibilityUpdateRequest(hidden=[]), api_request
|
|
)
|
|
assert exc.value.status_code == 403
|
|
|
|
|
|
# ── 定时任务「抽取信息写回模板数据」开关(html_config.template_data_writeback) ──
|
|
|
|
|
|
def _writeback_task(writeback: bool | None) -> dict[str, Any]:
|
|
ctx: dict[str, Any] = {"require_html": True, "html_config": {}}
|
|
if writeback is not None:
|
|
ctx["html_config"]["template_data_writeback"] = writeback
|
|
return {
|
|
"task_id": "t1",
|
|
"user_id": "u1",
|
|
"name": "回写联调",
|
|
"prompt": "更新总览的核心指标名称为「回写指标」",
|
|
"description": "",
|
|
"timezone": "UTC",
|
|
"enabled": True,
|
|
"execution_context": ctx,
|
|
}
|
|
|
|
|
|
def _build_writeback_service(monkeypatch, *, converted: dict | None):
|
|
"""构造 mock 掉检索/转换/归档的 service,返回 (service, write_mock)。"""
|
|
from datetime import UTC
|
|
from datetime import datetime as dt
|
|
from unittest.mock import AsyncMock, MagicMock
|
|
|
|
from deerflow.persistence.scheduled_tasks import MemoryScheduledTaskStore
|
|
from deerflow.runtime.scheduler import service as scheduler_service
|
|
from deerflow.runtime.scheduler.service import ScheduledTaskService
|
|
|
|
service = ScheduledTaskService(app=None, store=MemoryScheduledTaskStore(), poll_interval_seconds=60.0)
|
|
ref = template_fill.reference_data(TPL)
|
|
source_obj = json.loads(json.dumps(ref))
|
|
source_obj["overview"] = {"核心指标": [{"名称": "回写指标", "数值": 3, "单位": "个", "说明": ""}]}
|
|
monkeypatch.setattr(
|
|
service, "_gather_html_data", AsyncMock(return_value=("一些源数据", []))
|
|
)
|
|
if converted is None:
|
|
async def _fail(*a, **k):
|
|
raise scheduler_service.TemplateFillError("boom")
|
|
monkeypatch.setattr(service, "_convert_to_template_data", AsyncMock(side_effect=_fail))
|
|
else:
|
|
# 成功分支返回「抽取了新指标」的转换结果(converted 仅作成功/失败哨兵)。
|
|
monkeypatch.setattr(service, "_convert_to_template_data", AsyncMock(return_value=source_obj))
|
|
monkeypatch.setattr(service, "_finalize_template_html", AsyncMock(return_value={"ok": True}))
|
|
monkeypatch.setattr(service, "_archive_html_index", lambda *_a, **_k: None)
|
|
# write_template_data 是同步函数,用 MagicMock 而非 AsyncMock。
|
|
write_mock = MagicMock()
|
|
monkeypatch.setattr(template_store, "write_template_data", write_mock)
|
|
scheduled_for = dt(2026, 11, 4, 12, 0, tzinfo=UTC)
|
|
return service, write_mock, scheduled_for, source_obj
|
|
|
|
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
@pytest.mark.parametrize("writeback,converted,expect_write", [
|
|
(True, "converted", True),
|
|
(False, "converted", False),
|
|
(None, "converted", False), # 老任务缺省关闭
|
|
(True, None, False), # 转换失败(回退默认数据)不写回
|
|
])
|
|
async def test_html_template_reference_writeback_switch(tmp_store, monkeypatch, writeback, converted, expect_write):
|
|
service, write_mock, scheduled_for, _source = _build_writeback_service(
|
|
monkeypatch, converted=converted
|
|
)
|
|
await service._run_html_template_reference(
|
|
_writeback_task(writeback),
|
|
tpl_id=TPL,
|
|
runtime_thread_id="rt",
|
|
scheduled_for=scheduled_for,
|
|
run_id="r1",
|
|
)
|
|
if expect_write:
|
|
assert write_mock.call_count == 1
|
|
# 只写 data.json:调用参数不含 manifest kwarg(系统名称等 MANIFEST 不动)。
|
|
args, kwargs = write_mock.call_args
|
|
assert args[0] == TPL and "manifest" not in kwargs
|
|
assert args[1]["overview"]["核心指标"][0]["名称"] == "回写指标"
|
|
else:
|
|
assert write_mock.call_count == 0
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_writeback_respects_field_locks(tmp_store, monkeypatch):
|
|
"""锁定字段不接受抽取更新:写回 data.json 的数据里,锁定键保留模板旧值。"""
|
|
template_store.write_template_field_locks(TPL, ["日期"])
|
|
service, write_mock, scheduled_for, source_obj = _build_writeback_service(
|
|
monkeypatch, converted=source_marker()
|
|
)
|
|
await service._run_html_template_reference(
|
|
_writeback_task(True),
|
|
tpl_id=TPL,
|
|
runtime_thread_id="rt",
|
|
scheduled_for=scheduled_for,
|
|
run_id="r-lock",
|
|
)
|
|
assert write_mock.call_count == 1
|
|
args, _kwargs = write_mock.call_args
|
|
old_dates = [e.get("日期显示") for e in template_fill.reference_data(TPL)["events"]]
|
|
new_dates = [e.get("日期显示") for e in args[1]["events"]]
|
|
assert new_dates == old_dates # 锁定键恢复旧值
|
|
|
|
|
|
def source_marker():
|
|
ref = template_fill.reference_data(TPL)
|
|
obj = json.loads(json.dumps(ref))
|
|
for e in obj["events"]:
|
|
e["日期显示"] = "2099-12-31"
|
|
return obj
|