"""结构化 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