"""Architecture Lane Visualization + Calculation. Dataset + charts + Calculation IR + local evaluator. Cloud calculation.evaluate is the admitted authoritative extension; the local evaluator does not execute it. No Reporting reopen, no chart_definition / calculator_definition entities. """ from pathlib import Path import pytest import yaml from src.domain.app_blocks.canonical import CANONICAL_BLOCK_IDS, feature_pack_only_aliases from src.domain.app_blocks.catalogs import FEATURE_PACK_ALIASES, default_known_packs from src.domain.app_blocks.compiler import compile_install_plan from src.domain.app_blocks.execution import assert_install_plan_installable from src.domain.app_blocks.hosts import merge_host_map from src.domain.app_blocks.planner import ProductIntent, select_capabilities from src.domain.app_blocks.registry import load_default_registry from src.domain.calculation.errors import ( AuthoritativeNotAdmittedError, CalculationEvalError, ) from src.domain.calculation.evaluator import evaluate_local from src.domain.calculation.ir import CalculationIR from src.domain.calculation.mortgage import MORTGAGE_IR from src.domain.datasets.contract import DatasetContract from src.domain.feature_packs.normalize import load_pack_contract, stamp_composition from src.domain.starter_library.catalog import known_contract_ids from src.domain.visualization.chart_spec import CHART_TYPES_V1, ChartSpec from tests.test_phase11_pack_migration_materialization import _captured_composition ROOT = Path(__file__).resolve().parents[1] RECEIPT = ROOT / "src/domain/app_blocks/recipes/visualization_calculation_freeze.receipt.txt" AUDIT = ROOT / "docs/audits/ARCHITECTURE_LANE_VISUALIZATION_CALCULATION.md" REGISTRY = ROOT / "src/domain/app_blocks/platform_coverage_registry.v1.yaml" GATED = ("payments.checkout", "membership.portal") def _ids(intent: ProductIntent) -> set[str]: return {block_id for _cap, block_id, _reason in select_capabilities(intent)} def test_visualization_and_calculation_are_canonical_without_entities(): registry = load_default_registry() chart = registry.require("visualization.chart") calculator = registry.require("calculation.calculator") assert chart.status == "implemented" assert calculator.status == "implemented" assert {"visualization.chart", "calculation.calculator"} <= CANONICAL_BLOCK_IDS assert "visualization.chart" not in FEATURE_PACK_ALIASES assert "calculation.calculator" not in feature_pack_only_aliases() chart_c = registry.canonical.require("visualization.chart") calc_c = registry.canonical.require("calculation.calculator") assert chart_c.installation.mechanism == "none" assert calc_c.installation.mechanism == "none" assert chart_c.installation.feature_packs == () assert calc_c.installation.feature_packs == () primitives = {dep.id for dep in chart_c.requires if dep.kind == "primitive"} assert primitives == set() assert "calculator-mortgage" in default_known_packs() def test_mortgage_pack_ships_ir_without_an_entity(): contract = load_pack_contract("calculator-mortgage", {"entities": []}) assert contract.entities == [] assert "calculator.mortgage" in contract.provide_ids() assert [item.id for item in contract.calculators] == ["mortgage_calculator"] assert contract.calculators[0].execution == "local" stamped = stamp_composition("calculator-mortgage", {"entities": []}) assert stamped["entities"] in ([], None) or stamped.get("entities") == [] assert stamped["calculators"][0]["id"] == "mortgage_calculator" composition = _captured_composition( "alembic/versions/163_calculator_mortgage_pack.py", "calculator-mortgage" ) assert composition["entities"] == [] assert composition["calculators"][0]["id"] == "mortgage_calculator" assert "entity_contracts" in composition assert composition["entity_contracts"] == [] def test_chart_install_plan_does_not_require_reporting_or_a_pack(monkeypatch): from src.domain.app_blocks import compiler registry = load_default_registry() monkeypatch.setattr( compiler, "select_capabilities", lambda _intent: [("visualization_chart", "visualization.chart", "Selected.")], ) plan = compiler.compile_install_plan(None, registry=registry) assert_install_plan_installable(plan) assert "reporting-dashboards" not in plan.required_feature_packs assert "reporting.dashboard" not in {block.id for block in plan.blocks} assert plan.required_feature_packs == [] def test_mortgage_prompt_installs_calculator_pack_not_reporting(): prompt = "Add a mortgage calculator to estimate monthly loan payments." ids = _ids(ProductIntent(purpose=prompt, source_prompt=prompt)) assert "calculation.calculator" in ids assert "reporting.dashboard" not in ids assert not (ids & set(GATED)) plan = compile_install_plan( { "product": {"purpose": prompt, "first_build_goal": prompt}, "original_intent": prompt, "actors": [{"id": "staff", "name": "Staff"}], "workflows": [{"id": "review", "name": "Review", "description": "Review"}], "entities": [{"id": "record", "name": "Record", "description": "A record"}], "tenancy": {"model": "single_workspace"}, "authentication": {"required": False}, }, source_prompt=prompt, ) assert "calculator-mortgage" in plan.required_feature_packs assert "reporting-dashboards" not in plan.required_feature_packs def test_bar_chart_does_not_select_reporting(): prompt = "Show a bar chart of monthly revenue from existing records." ids = _ids(ProductIntent(purpose=prompt, source_prompt=prompt)) assert "visualization.chart" in ids assert "reporting.dashboard" not in ids assert "billing.invoice" not in ids def test_reporting_still_does_not_select_charts(): prompt = "Staff need a reporting dashboard." ids = _ids(ProductIntent(purpose=prompt, source_prompt=prompt)) assert "reporting.dashboard" in ids assert "visualization.chart" not in ids def test_false_positives_do_not_select_charts_or_calculators(): for purpose in ( "chart of accounts", "org chart for the firm", "Staff manage invoices. No reports or dashboards.", "calculate payroll later", "We use Tableau for that already", "pricing page", ): ids = _ids(ProductIntent(purpose=purpose, source_prompt=purpose)) assert "visualization.chart" not in ids, purpose assert "calculation.calculator" not in ids, purpose def test_negative_intent_skips_charts_and_calculators(): for prompt in ( "Client records. Do not add a calculator.", "Staff CRM. No charts.", ): ids = _ids(ProductIntent(purpose=prompt, source_prompt=prompt)) assert "visualization.chart" not in ids, prompt assert "calculation.calculator" not in ids, prompt def test_codegen_margin_calculator_selects_the_block_without_mortgage_pack(): prompt = "Add a calculator that estimates consulting project margin." ids = _ids(ProductIntent(purpose=prompt, source_prompt=prompt)) assert "calculation.calculator" in ids plan = compile_install_plan( { "product": {"purpose": prompt, "first_build_goal": prompt}, "original_intent": prompt, "actors": [{"id": "staff", "name": "Staff"}], "workflows": [{"id": "review", "name": "Review", "description": "Review"}], "entities": [{"id": "record", "name": "Record", "description": "A record"}], "tenancy": {"model": "single_workspace"}, "authentication": {"required": False}, }, source_prompt=prompt, ) assert "calculator-mortgage" not in plan.required_feature_packs def test_starter_presentation_ids_are_admitted_and_distinct_from_reporting_metrics(): known = known_contract_ids() for contract_id in ( "chart.bar", "chart.line", "chart.pie", "chart.area", "metric.card", "calculator.formula", ): assert contract_id in known assert "stat.card" not in known registry = load_default_registry() chart_slots = registry.require("visualization.chart").slots calc_slots = registry.require("calculation.calculator").slots chart = {slot.id: slot.component_contract for slot in chart_slots} calc = {slot.id: slot.component_contract for slot in calc_slots} assert chart["chart"] == "chart.bar" assert chart["metric"] == "metric.card" assert calc["formula"] == "calculator.formula" assert calc["result"] == "metric.card" reporting = registry.require("reporting.dashboard") assert "metric.count" in {item.id for item in reporting.provides} assert "metric.card" not in {item.id for item in reporting.provides} def test_dataset_contract_is_project_scoped_and_bounded(): dataset = DatasetContract.model_validate( { "id": "revenue_by_month", "source": {"kind": "records.aggregate", "entity_key": "invoice_record"}, "shape": {"kind": "series", "fields": ["month", "revenue"]}, "scope": {"project": "current", "filters": {"status": "paid"}}, "audience": "authenticated", "freshness": {"mode": "live"}, "row_budget": {"max": 1000}, } ) assert dataset.scope.project == "current" with pytest.raises(Exception): DatasetContract.model_validate( { "id": "cross", "source": {"kind": "static", "static_rows": []}, "shape": {"kind": "series"}, "scope": {"project": "other"}, } ) def test_chart_spec_v1_types_only(): spec = ChartSpec.model_validate( { "id": "revenue_chart", "variant": "bar", "dataset": "revenue_by_month", "encoding": {"x": "month", "y": "revenue"}, } ) assert spec.starter_contract == "chart.bar" assert spec.variant in CHART_TYPES_V1 def test_local_mortgage_evaluator_and_dataset_output(): result = evaluate_local( MORTGAGE_IR, inputs={"principal": "250000", "annual_rate": "0.065", "years": "30"}, ) payment = float(result["outputs"]["monthly_payment"]) assert 1500 < payment < 1700 assert result["outputs"]["summary"]["shape"] == "series" assert len(result["outputs"]["summary"]["rows"]) == 3 def test_authoritative_ir_is_valid_but_requires_cloud_execution(): payload = MORTGAGE_IR.model_dump(by_alias=True) payload["id"] = "invoice_total" payload["execution"] = "authoritative" ir = CalculationIR.model_validate(payload) assert ir.execution == "authoritative" with pytest.raises(AuthoritativeNotAdmittedError): evaluate_local(ir, inputs={"principal": 1, "annual_rate": 0, "years": 1}) def test_ir_cannot_query_records_and_rejects_cycles(): with pytest.raises(Exception): CalculationIR.model_validate( { "id": "bad", "inputs": [], "fields": [{"key": "x", "expr": {"op": "dataset", "id": "customers"}}], "outputs": [{"key": "x", "from_field": "x"}], } ) with pytest.raises(Exception, match="cycle"): CalculationIR.model_validate( { "id": "cycle", "inputs": [{"key": "a", "type": "number"}], "fields": [ {"key": "x", "expr": {"op": "field", "key": "y"}}, {"key": "y", "expr": {"op": "field", "key": "x"}}, ], "outputs": [{"key": "x", "from_field": "x"}], } ) def test_local_eval_uses_dataset_snapshot_not_a_query(): ir = CalculationIR.model_validate( { "id": "savings", "execution": "local", "inputs": [], "datasets": [{"id": "usage", "dataset_id": "usage"}], "fields": [ { "key": "total", "expr": {"op": "sum", "over": {"op": "dataset", "id": "usage"}, "field": "kwh"}, } ], "outputs": [{"key": "total", "from_field": "total"}], } ) result = evaluate_local(ir, datasets={"usage": [{"kwh": 10}, {"kwh": 15}]}) assert result["outputs"]["total"] in {"25", "25.0"} with pytest.raises(CalculationEvalError): evaluate_local(ir, datasets={}) def test_local_eval_can_order_bounded_pick_rows_with_missing_values_last(): ir = CalculationIR.model_validate( { "id": "pick-list-order", "datasets": [{"id": "picks", "dataset_id": "picks"}], "fields": [ { "key": "ordered", "expr": { "op": "sort", "over": {"op": "dataset", "id": "picks"}, "field": "location", }, } ], "outputs": [{"key": "ordered", "kind": "dataset", "shape": "table", "from_field": "ordered"}], } ) result = evaluate_local( ir, datasets={"picks": [{"sku": "B", "location": "B-02"}, {"sku": "A", "location": "A-01"}, {"sku": "C"}]}, ) assert [row["sku"] for row in result["outputs"]["ordered"]["rows"]] == ["A", "B", "C"] def test_hosts_exist_and_do_not_invent_a_bi_or_cloud_evaluate_api(): import json for framework in ("html", "nuxt_app", "react_app"): root = ROOT / "boilerplates" / framework hosts = merge_host_map(json.loads((root / "p2hs/app-block-hosts.json").read_text())) charts = {h["slot"]: h for h in hosts["hosts"] if h["block"] == "visualization.chart"} calcs = {h["slot"]: h for h in hosts["hosts"] if h["block"] == "calculation.calculator"} assert {"chart", "metric", "empty"} <= set(charts) assert {"formula", "result", "empty"} <= set(calcs) for group in (charts, calcs): for slot in group.values(): path = slot["implementation"].partition("#")[0] source = (root / path).read_text(encoding="utf-8") assert "chart_definition" not in source assert "calculator_definition" not in source assert "/api/reporting" not in source assert "calculation.evaluate" not in source def test_visualization_calculation_freeze_receipts(): receipt = RECEIPT.read_text(encoding="utf-8") audit = AUDIT.read_text(encoding="utf-8") for required in ( "visualization_calculation_admitted=yes", "canonical_app_block=visualization.chart", "canonical_app_block=calculation.calculator", "chart_definition_entity=none", "calculator_definition_entity=none", "calculation.evaluate=admitted", "reporting_dependency=none", "family_frozen=yes", ): assert required in receipt, required assert "visualization_calculation_freeze=yes" in audit registry = yaml.safe_load(REGISTRY.read_text(encoding="utf-8")) by_id = {row["id"]: row for row in registry["capabilities"]} assert registry["family_just_frozen"] == "visualization_calculation" assert by_id["visualization.chart"]["status"] == "frozen" assert by_id["calculation.calculator"]["status"] == "frozen" assert by_id["calculator-mortgage"]["status"] == "frozen" assert by_id["calculation.evaluate"]["status"] == "admitted" assert by_id["reporting.dashboard"]["status"] == "frozen" def test_visualization_calculation_production_smoke_script(): from scripts.architecture_lane_visualization_calculation_smoke import run_smoke payload = run_smoke() assert payload["pass"] is True, payload["failures"]