wsb1224
2422303b36
阶段 当前状态 说明 Phase 0～3 基本完成 Document IR、证据链、人工确认、不可变 PlanData Snapshot、海报/PPT 投影已实现 Phase 4 代码完成 场景、策略、模板版本，校验/发布门禁，确定性 resolver 和严格槽位合并已实现 Phase 5 代码完成 输入冻结、PPTX/海报对账、失败关闭、幂等、心跳、重试和冻结输入重放已实现 Phase 6 基础完成 质量看板、结构化告警、Golden 审批、留存清理、孤儿检查和灰度开关已实现 正式上线 未完成 缺真实样本、生产模板、业务规则签字和灰度观察 目前验证基线： PPT/海报专项测试：107 passed, 1 skipped Vue TypeScript 检查：通过 前端生产构建：通过 代码变更仍在工作区，尚未提交 仓库全量测试仍有既有失败/挂起项，暂时不能宣称全仓测试完全绿色 仍未完成的代码任务主要有： 影子解析差异流水线 目前有灰度开关，但还没有完整的“新旧解析同时运行、字段差异入库、按保司/profile 聚合”的影子比较任务。 自动视觉回归 目前实现的是 DOM 模块、溢出、尺寸、文本和数值检查；还缺基于真实模板和标准图片的像素差异、字体缺失、遮挡和裁切回归。 告警通道接入 后台已经能产生结构化质量告警，但尚未自动推送到邮件、企微或其他通知通道。 留存任务生产化 dry-run、实删服务和失败审计已经具备，但尚未接入周期性 Celery/定时任务，也没有自动重试失败清理批次。 旧链路最终下线 旧 PPT 解析器和海报紧凑解析仍保留为回滚路径。需要全量灰度稳定后才能删除或彻底关闭写入口。 全仓测试收口 需要处理现有无关失败和挂起测试，建立真正全绿的 CI 基线。 仍需外部输入和生产环境完成的事项： 至少 30 份脱敏 Golden PDF，并完成双人标注和精确率验收。 业务专家确认派生公式、缺失值、可比较性和结论策略。 上传并标注真实生产 PPTX 的语义 shape、页面类型和容量。 安装生产字体并建立视觉基准图片。 实际执行数据库迁移 038～040。 完成留存 dry-run、实删演练以及 10% → 30% → 100% 灰度。 观察期通过后开启 SCENARIO_ENGINE_V2，目前它仍默认关闭；真实清理开关也默认关闭。 完整状态记录在 [PPT与海报Phase4至6补充实施记录](D:/work/code/python/coding/baodanagent/docs/PPT与海报Phase4至6补充实施记录_20260802.md)。
282 lines
12 KiB
Python
282 lines
12 KiB
Python
import sys
|
|
from datetime import datetime, timedelta
|
|
from pathlib import Path
|
|
|
|
import pytest
|
|
from flask import Flask
|
|
|
|
|
|
sys.path.insert(0, str(Path(__file__).resolve().parents[1] / "api"))
|
|
|
|
|
|
def _app():
|
|
from insurance.db.database import db
|
|
import insurance.models # noqa: F401
|
|
|
|
app = Flask(__name__)
|
|
app.config.update(
|
|
SQLALCHEMY_DATABASE_URI="sqlite:///:memory:",
|
|
SQLALCHEMY_TRACK_MODIFICATIONS=False,
|
|
)
|
|
db.init_app(app)
|
|
return app
|
|
|
|
|
|
def _seed_snapshot():
|
|
from insurance.db.compat import db
|
|
from insurance.models.insurance_document import InsuranceDocument
|
|
from insurance.models.plan_snapshot import InsurancePlanSnapshot
|
|
|
|
document = InsuranceDocument(
|
|
user_id="user-1", sha256="a" * 64, storage_key="plans/a.pdf",
|
|
original_name="a.pdf", mime_type="application/pdf", file_size=10,
|
|
page_count=1, pdf_kind="native", status="parsed", parser_version="test",
|
|
)
|
|
db.session.add(document)
|
|
db.session.flush()
|
|
snapshot = InsurancePlanSnapshot(
|
|
user_id="user-1", document_id=document.id, snapshot_version=1,
|
|
status="confirmed", plan_data_json="{}", validation_results_json='{"valid":true}',
|
|
snapshot_hash="b" * 64, document_sha256=document.sha256,
|
|
confirmed_by="user-1", confirmed_at=datetime.now(),
|
|
is_golden=True, golden_approved_by="admin", golden_approved_at=datetime.now(),
|
|
)
|
|
db.session.add(snapshot)
|
|
db.session.commit()
|
|
return snapshot
|
|
|
|
|
|
def _scenario_definition(priority=10):
|
|
return {
|
|
"selector": {"priority": priority, "fileCount": 1, "planTypes": ["savings"]},
|
|
"pageSpecs": [{"pageType": "cover"}, {"pageType": "chart"}],
|
|
"metricSpecs": [{
|
|
"code": "surrender_value", "fieldPath": "benefitRows[].totalSurrenderValue",
|
|
"pageTypes": ["chart"], "required": True,
|
|
}],
|
|
"comparisonYears": [10, 20, 30],
|
|
"compatibilityRules": {"planTypes": ["savings"]},
|
|
}
|
|
|
|
|
|
def test_version_lifecycle_publish_gate_and_deterministic_multi_match():
|
|
from insurance.db.compat import db
|
|
from insurance.models.ppt_config import PptScenario
|
|
from insurance.ppt.versioning import (
|
|
VersioningError, create_scenario_version, publish_scenario_version,
|
|
resolve_published_scenario, validate_scenario_version,
|
|
)
|
|
|
|
app = _app()
|
|
with app.app_context():
|
|
db.create_all()
|
|
snapshot = _seed_snapshot()
|
|
for code in ("retirement", "wealth"):
|
|
db.session.add(PptScenario(code=code, name=code, generation_mode="single"))
|
|
db.session.commit()
|
|
|
|
first = create_scenario_version("retirement", _scenario_definition(), "admin")
|
|
with pytest.raises(VersioningError) as no_golden:
|
|
validate_scenario_version(first.id, {})
|
|
assert no_golden.value.code == "VALIDATION_FAILED"
|
|
|
|
first = validate_scenario_version(first.id, {"sampleSnapshotIds": [snapshot.id]})
|
|
assert first.lifecycle == "validated"
|
|
first = publish_scenario_version(first.id, "admin")
|
|
assert first.lifecycle == "published"
|
|
|
|
second = create_scenario_version("wealth", _scenario_definition(), "admin")
|
|
validate_scenario_version(second.id, {"sampleSnapshotIds": [snapshot.id]})
|
|
publish_scenario_version(second.id, "admin")
|
|
|
|
with pytest.raises(VersioningError) as ambiguous:
|
|
resolve_published_scenario({"fileCount": 1, "planTypes": ["savings"]})
|
|
assert ambiguous.value.code == "SCENARIO_MULTI_MATCH"
|
|
|
|
|
|
def test_template_and_policy_versions_publish_only_after_validation():
|
|
from insurance.db.compat import db
|
|
from insurance.models.ppt_config import PptScenario, PptTemplate
|
|
from insurance.ppt.versioning import (
|
|
_inspect_pptx_asset,
|
|
create_policy_version, create_scenario_version, create_template_version,
|
|
publish_policy_version, publish_scenario_version, publish_template_version,
|
|
validate_policy_version, validate_scenario_version, validate_template_version,
|
|
)
|
|
|
|
app = _app()
|
|
with app.app_context():
|
|
db.create_all()
|
|
snapshot = _seed_snapshot()
|
|
from insurance.ppt.template_asset_service import resolve_template_asset, template_asset_sha256
|
|
asset_id = "builtin://single_savings.pptx"
|
|
asset_path = resolve_template_asset(asset_id)
|
|
asset_hash = template_asset_sha256(asset_path)
|
|
shape_names, _ = _inspect_pptx_asset(asset_path)
|
|
cover_shape = sorted(shape_names[0])[0]
|
|
chart_shape = sorted(shape_names[1])[0]
|
|
db.session.add(PptScenario(code="single", name="single", generation_mode="single"))
|
|
db.session.add(PptTemplate(
|
|
id="broker", plan_type="savings", style_preset="broker",
|
|
source_template_asset_id=asset_id, asset_sha256=asset_hash,
|
|
required_page_types_json='["cover"]',
|
|
))
|
|
db.session.commit()
|
|
scenario = create_scenario_version("single", _scenario_definition(), "admin")
|
|
validate_scenario_version(scenario.id, {"sampleSnapshotIds": [snapshot.id]})
|
|
publish_scenario_version(scenario.id, "admin")
|
|
|
|
policy = create_policy_version("default", {
|
|
"calculationPolicy": {"totals": "contract_only", "derivedMetrics": []},
|
|
"missingValuePolicy": {"mode": "block"},
|
|
"conclusionPolicy": {"mode": "rules_only"},
|
|
}, "admin")
|
|
validate_policy_version(policy.id)
|
|
assert publish_policy_version(policy.id, "admin").lifecycle == "published"
|
|
|
|
template = create_template_version("broker", {
|
|
"pageSlots": [{
|
|
"slotId": "cover-1", "pageType": "cover", "sourceSlideIndex": 1,
|
|
"shapeSlots": [{"shapeName": cover_shape, "role": "title"}],
|
|
}, {
|
|
"slotId": "chart-1", "pageType": "chart", "sourceSlideIndex": 2,
|
|
"shapeSlots": [{"shapeName": chart_shape, "role": "chart"}],
|
|
}],
|
|
"capacityRules": {"maxTitleChars": 40},
|
|
"supportedScenarioVersionIds": [scenario.id],
|
|
}, "admin")
|
|
validate_template_version(template.id, {"sampleSnapshotIds": [snapshot.id]})
|
|
assert publish_template_version(template.id, "admin").lifecycle == "published"
|
|
|
|
|
|
def test_retention_dry_run_records_audit_without_deleting(monkeypatch):
|
|
from insurance.db.compat import db
|
|
from insurance.models.insurance_document import InsuranceDocument
|
|
from insurance.models.retention_audit import RetentionCleanupAudit
|
|
from insurance.retention.service import run_cleanup
|
|
|
|
app = _app()
|
|
with app.app_context():
|
|
db.create_all()
|
|
monkeypatch.setenv("INSURANCE_DOCUMENT_RETENTION_DAYS", "30")
|
|
document = InsuranceDocument(
|
|
user_id="user-1", sha256="d" * 64, storage_key="plans/expired.pdf",
|
|
original_name="expired.pdf", mime_type="application/pdf", file_size=10,
|
|
page_count=1, pdf_kind="native", status="parsed", parser_version="test",
|
|
expires_at=datetime.now() - timedelta(days=1),
|
|
)
|
|
db.session.add(document)
|
|
db.session.commit()
|
|
document_id = document.id
|
|
|
|
result = run_cleanup(dry_run=True)
|
|
|
|
assert result["count"] == 1
|
|
assert db.session.get(InsuranceDocument, document_id) is not None
|
|
assert RetentionCleanupAudit.query.one().action == "dry_run"
|
|
|
|
|
|
def test_retention_orphan_scan_is_read_only_and_uses_relative_paths(monkeypatch, tmp_path):
|
|
from insurance.db.compat import db
|
|
from insurance.models.insurance_document import InsuranceDocument
|
|
from insurance.retention.service import find_orphan_files
|
|
|
|
app = _app()
|
|
with app.app_context():
|
|
db.create_all()
|
|
monkeypatch.setenv("INSURANCE_STORAGE_ROOT", str(tmp_path))
|
|
referenced = tmp_path / "plans" / "kept.pdf"
|
|
orphan = tmp_path / "outputs" / "unused.png"
|
|
referenced.parent.mkdir(parents=True)
|
|
orphan.parent.mkdir(parents=True)
|
|
referenced.write_bytes(b"pdf")
|
|
orphan.write_bytes(b"png")
|
|
db.session.add(InsuranceDocument(
|
|
user_id="user-1", sha256="e" * 64, storage_key="plans/kept.pdf",
|
|
original_name="kept.pdf", mime_type="application/pdf", file_size=3,
|
|
page_count=1, pdf_kind="native", status="parsed", parser_version="test",
|
|
))
|
|
db.session.commit()
|
|
|
|
result = find_orphan_files()
|
|
|
|
assert result["orphanCount"] == 1
|
|
assert result["items"][0]["relativePath"] == "outputs/unused.png"
|
|
assert orphan.exists()
|
|
|
|
|
|
def test_runtime_merge_is_exact_and_capacity_fails_closed():
|
|
from insurance.ppt.versioning import VersioningError, build_reconciliation_manifest, build_runtime_merge
|
|
|
|
scenario = {"pageSpecs": [
|
|
{"pageType": "cover", "title": "方案"},
|
|
{"pageType": "chart", "title": "利益趋势"},
|
|
]}
|
|
template = {
|
|
"capacityRules": {"maxTitleChars": 8},
|
|
"pageSlots": [
|
|
{"slotId": "chart", "pageType": "chart", "sourceSlideIndex": 4,
|
|
"shapeSlots": [{"shapeName": "Chart", "role": "chart"}]},
|
|
{"slotId": "cover", "pageType": "cover", "sourceSlideIndex": 1,
|
|
"shapeSlots": [{"shapeName": "Title", "role": "title"}]},
|
|
],
|
|
}
|
|
result = build_runtime_merge(scenario, template)
|
|
assert result["templateConfig"]["frameMap"] == [1, 4]
|
|
assert result["templateConfig"]["businessShapeNamesBySlide"] == [["Title"], ["Chart"]]
|
|
|
|
with pytest.raises(VersioningError) as overflow:
|
|
build_runtime_merge({"pageSpecs": [{"pageType": "cover", "title": "超过容量的标题文字"}]}, template)
|
|
assert overflow.value.code == "TEMPLATE_CAPACITY_EXCEEDED"
|
|
|
|
with pytest.raises(VersioningError) as missing:
|
|
build_reconciliation_manifest(
|
|
[{"policy": {"currency": "HKD"}}],
|
|
{"comparisonYears": [10], "metricSpecs": [{
|
|
"fieldPath": "policy.annualPremium", "pageTypes": ["chart"], "required": True,
|
|
}]},
|
|
{"missingValuePolicy": {"mode": "block"}},
|
|
)
|
|
assert missing.value.code == "REQUIRED_METRIC_MISSING"
|
|
|
|
|
|
def test_replay_reuses_frozen_input(monkeypatch):
|
|
from insurance.db.compat import db
|
|
from insurance.generation import task_service
|
|
from insurance.models.generation_task import GenerationTask
|
|
|
|
app = _app()
|
|
with app.app_context():
|
|
db.create_all()
|
|
original = GenerationTask(
|
|
id="old-task", user_id="user-1", artifact_type="ppt", operation="generate",
|
|
workspace_id="workspace-1", status="failed", input_revision=3, submit_revision=3,
|
|
input_snapshot_json='{"snapshotHash":"abc","deckContracts":[{"kind":"savings"}]}'
|
|
)
|
|
db.session.add(original)
|
|
db.session.commit()
|
|
captured = {}
|
|
|
|
def fake_create_task(**kwargs):
|
|
captured.update(kwargs)
|
|
return {"code": 0, "data": {"id": "new-task"}}
|
|
|
|
monkeypatch.setattr(task_service, "create_task", fake_create_task)
|
|
result = task_service.replay_task(original.id, "user-1")
|
|
|
|
assert result["code"] == 0
|
|
assert captured["input_snapshot"]["deckContracts"] == [{"kind": "savings"}]
|
|
assert captured["input_snapshot"]["replay"]["sourceTaskId"] == "old-task"
|
|
assert captured["input_revision"] == 3
|
|
|
|
|
|
def test_feature_flag_rollout_is_stable_and_profile_scoped(monkeypatch):
|
|
from insurance.generation.feature_flags import enabled_for
|
|
|
|
monkeypatch.setenv("INSURANCE_SCENARIO_ENGINE_V2", "true")
|
|
monkeypatch.setenv("INSURANCE_SCENARIO_ENGINE_V2_PERCENT", "50")
|
|
monkeypatch.setenv("INSURANCE_SCENARIO_ENGINE_V2_PROFILES", "company-a")
|
|
first = enabled_for("SCENARIO_ENGINE_V2", identity="user-1", profile="company-a")
|
|
assert enabled_for("SCENARIO_ENGINE_V2", identity="user-1", profile="company-a") is first
|
|
assert enabled_for("SCENARIO_ENGINE_V2", identity="user-1", profile="company-b") is False
|