baodan/tests/plan_snapshot_test.py

import json
import sys
from pathlib import Path

import pytest


sys.path.insert(0, str(Path(__file__).resolve().parents[1] / "api"))


def test_generate_ppt_does_not_shadow_module_jsonify():
    import symtable

    route_path = Path(__file__).resolve().parents[1] / "api/insurance/ppt/routes.py"
    module = symtable.symtable(route_path.read_text(encoding="utf-8"), str(route_path), "exec")
    generate_ppt = next(child for child in module.get_children() if child.get_name() == "generate_ppt")
    jsonify = generate_ppt.lookup("jsonify")

    assert jsonify.is_global()
    assert not jsonify.is_local()


def _evidence(document_id=1, page=2):
    return {
        "documentId": document_id,
        "pageNumber": page,
        "bbox": [10, 20, 100, 40],
        "tableId": None,
        "rowId": None,
        "columnId": None,
        "textHash": "a" * 64,
    }


def _legacy_data(document_id=1):
    return {
        "product_name": "储蓄计划 A",
        "product_type": "savings",
        "insured": {"age": 35, "gender": "male", "smoker": "no"},
        "policy": {
            "currency": "HKD",
            "annual_premium": 100000,
            "premium_payment_period": 5,
            "total_premium": None,
        },
        "benefit_illustration": [{
            "policy_year": 10,
            "scenario_type": "base",
            "total_surrender_value": 650000,
        }],
        "meta": {"evidence": [
            {"fieldPath": "policy.annual_premium", "evidence": _evidence(document_id, 2)},
            {"fieldPath": "benefit.base.10.default.total_surrender_value", "evidence": _evidence(document_id, 8)},
        ]},
    }


def test_candidate_reconciliation_preserves_conflicts_and_context_hint_cannot_override():
    from insurance.plan_data.field_values import make_candidate, reconcile_candidates

    evidence = [_evidence()]
    result = reconcile_candidates([
        make_candidate("policy.annualPremium", "100,000", normalized_value=100000, source="pdf_table", evidence=evidence),
        make_candidate("policy.annualPremium", 120000, source="llm", confidence=0.99),
        make_candidate("identity.productName", "文件名产品", source="context_hint"),
        make_candidate("identity.productName", "PDF 产品", source="regex", evidence=evidence),
    ])

    assert result["policy.annualPremium"]["status"] == "conflict"
    assert result["policy.annualPremium"]["conflictCandidates"] == [100000, 120000]
    assert result["identity.productName"]["status"] == "conflict"
    assert set(result["identity.productName"]["conflictCandidates"]) == {"PDF 产品", "文件名产品"}


def test_legacy_conversion_uses_field_values_and_never_derives_contract_total():
    from insurance.plan_data.conversion import legacy_to_plan_data

    plan_data = legacy_to_plan_data(_legacy_data())

    assert plan_data["policy"]["annualPremium"]["value"] == 100000
    assert plan_data["policy"]["annualPremium"]["status"] == "extracted"
    assert plan_data["policy"]["contractualTotalPremium"]["status"] == "missing"
    assert plan_data["policy"]["contractualTotalPremium"]["value"] is None
    assert plan_data["benefitScenarios"][0]["rows"][0]["totalSurrenderValue"]["evidence"]


def test_poster_flat_fields_survive_snapshot_conversion():
    from insurance.plan_data.conversion import legacy_to_plan_data
    from insurance.plan_data.snapshot_validation import validate_plan_data

    poster_data = {
        "product_name": "测试重疾计划",
        "plan_type": "ci",
        "age": 38,
        "gender": "female",
        "smoking_status": "no",
        "currency": "HKD",
        "sum_assured": 1_000_000,
        "annual_premium": 25_000,
        "premium_term": 20,
        "meta": {"evidence": []},
    }

    plan_data = legacy_to_plan_data(poster_data)
    result = validate_plan_data(plan_data, for_confirmation=True)

    assert plan_data["identity"]["insuredAge"]["value"] == 38
    assert plan_data["identity"]["insuredGender"]["value"] == "female"
    assert plan_data["identity"]["insuredSmoker"]["value"] == "no"
    assert plan_data["policy"]["currency"]["value"] == "HKD"
    assert plan_data["policy"]["sumAssured"]["value"] == 1_000_000
    assert plan_data["policy"]["premiumPaymentPeriod"]["value"] == 20
    assert result["blockingCount"] == 0


def test_projection_uses_confirmed_table_milestones_and_keeps_unknown_null():
    from insurance.plan_data.conversion import legacy_to_plan_data
    from insurance.plan_data.projections import to_deck_contract, to_poster_projection

    plan_data = legacy_to_plan_data(_legacy_data())
    poster = to_poster_projection(plan_data)
    deck = to_deck_contract(plan_data)

    assert poster["surrender_value_10"] == 650000
    assert poster["surrender_value_20"] is None
    assert poster["total_premium"] is None
    assert deck["policy"]["contractualTotalPremium"] is None
    assert deck["benefitRows"][0]["totalSurrenderValue"] == 650000


def test_migration_037_and_snapshot_lifecycle_are_versioned_and_owned():
    from flask import Flask
    from sqlalchemy import inspect

    from insurance.db.database import db
    from insurance.db.migrate_036 import migrate as migrate_036
    from insurance.db.migrate_037 import migrate as migrate_037
    from insurance.models.insurance_document import InsuranceDocument
    from insurance.plan_data.service import (
        SnapshotError,
        confirm_snapshot,
        create_draft_from_legacy,
        patch_draft,
    )

    app = Flask(__name__)
    app.config.update(
        SQLALCHEMY_DATABASE_URI="sqlite:///:memory:",
        SQLALCHEMY_TRACK_MODIFICATIONS=False,
    )
    db.init_app(app)
    with app.app_context():
        migrate_036()
        migrate_037()
        document = InsuranceDocument(
            user_id="user-1",
            sha256="b" * 64,
            storage_key="/private/plan.pdf",
            original_name="plan.pdf",
            mime_type="application/pdf",
            file_size=100,
            page_count=8,
            pdf_kind="native",
            status="parsed",
            parser_version="test-v1",
        )
        db.session.add(document)
        db.session.commit()
        legacy = _legacy_data(document.id)
        draft = create_draft_from_legacy(document.id, "user-1", legacy)
        confirmed = confirm_snapshot(draft.id, "user-1")

        assert draft.snapshot_version == 1
        assert confirmed.snapshot_version == 2
        assert confirmed.previous_snapshot_id == draft.id
        assert confirmed.status == "confirmed"
        assert confirmed.snapshot_hash != draft.snapshot_hash
        assert confirmed.validation()["blockingCount"] == 0

        with pytest.raises(SnapshotError) as stale:
            patch_draft(
                confirmed.id,
                "user-1",
                expected_hash="wrong",
                changes=[{
                    "fieldPath": "policy.annualPremium",
                    "value": 110000,
                    "overrideReason": "客户确认修正",
                }],
            )
        assert stale.value.code == "SNAPSHOT_VERSION_CONFLICT"

        edited = patch_draft(
            confirmed.id,
            "user-1",
            expected_hash=confirmed.snapshot_hash,
            changes=[{
                "fieldPath": "policy.annualPremium",
                "value": 110000,
                "overrideReason": "客户确认修正",
            }],
        )
        assert edited.snapshot_version == 3
        assert edited.plan_data()["policy"]["annualPremium"]["status"] == "confirmed"
        assert confirmed.plan_data()["policy"]["annualPremium"]["value"] == 100000

        with pytest.raises(SnapshotError):
            confirm_snapshot(edited.id, "user-2")

        tables = set(inspect(db.engine).get_table_names())
        assert {
            "insurance_plan_snapshots",
            "insurance_field_evidence",
            "insurance_plan_overrides",
        } <= tables


def test_confirmation_warns_when_critical_amount_evidence_is_missing():
    from insurance.plan_data.conversion import legacy_to_plan_data
    from insurance.plan_data.snapshot_validation import validate_plan_data

    data = _legacy_data()
    data["meta"]["evidence"] = []
    result = validate_plan_data(legacy_to_plan_data(data), for_confirmation=True)

    assert result["blockingCount"] == 0
    assert result["warningCount"] == 1
    assert any(item["code"] == "CRITICAL_EVIDENCE_MISSING" for item in result["issues"])


def test_optional_benefit_and_withdrawal_data_do_not_block_confirmation():
    from insurance.plan_data.conversion import legacy_to_plan_data
    from insurance.plan_data.snapshot_validation import validate_plan_data

    data = _legacy_data()
    data["meta"]["evidence"] = [
        {"fieldPath": "policy.annual_premium", "evidence": _evidence(1, 2)},
    ]
    data["withdrawal_illustration"] = [{
        "policy_year": 10,
        "annual_withdrawal": 50000,
        "surrender_value_after": 300000,
    }]
    result = validate_plan_data(legacy_to_plan_data(data), for_confirmation=True)

    assert result["blockingCount"] == 0
    assert not any(
        item["code"] == "CRITICAL_EVIDENCE_MISSING"
        and (
            ".benefitScenarios." in item["path"]
            or ".withdrawals." in item["path"]
        )
        for item in result["issues"]
    )


def test_missing_optional_projection_sections_do_not_block_confirmation():
    from insurance.plan_data.conversion import legacy_to_plan_data
    from insurance.plan_data.snapshot_validation import validate_plan_data

    data = _legacy_data()
    data["benefit_illustration"] = []
    data["withdrawal_illustration"] = []
    result = validate_plan_data(legacy_to_plan_data(data), for_confirmation=True)

    assert result["blockingCount"] == 0


def test_confirmation_does_not_treat_source_labels_as_amounts():
    from insurance.plan_data.conversion import legacy_to_plan_data
    from insurance.plan_data.snapshot_validation import validate_plan_data

    data = _legacy_data()
    data["policy"]["annual_premium_source_label"] = "IUL利益表保费列"
    result = validate_plan_data(legacy_to_plan_data(data), for_confirmation=True)

    assert not any(
        item["path"].endswith("annualPremiumSourceLabel")
        for item in result["issues"]
    )


def test_generation_task_exposes_frozen_versions_without_server_paths():
    from insurance.models.generation_task import GenerationTask

    task = GenerationTask(
        id="task-1",
        user_id="user-1",
        artifact_type="ppt",
        operation="generate",
        workspace_id="session-1",
        input_revision=7,
        submit_revision=7,
        snapshot_id=3,
        snapshot_hash="c" * 64,
        renderer_version="renderer-v2",
        result_state="stale",
        output_json=json.dumps({"filePath": "/private/output.pptx", "downloadUrl": "/download/1"}),
    )

    payload = task.to_dict()
    assert payload["snapshotId"] == 3
    assert payload["submitRevision"] == 7
    assert payload["rendererVersion"] == "renderer-v2"
    assert payload["resultState"] == "stale"
    assert payload["output"] == {"downloadUrl": "/download/1"}