"""Architecture Lane Reporting: admit report_definition over live Cloud aggregate ports. No new kernel, BI warehouse, cross-project analytics, or scheduled export pipeline. """ 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.execution import assert_install_plan_installable from src.domain.app_blocks.generated_product_conformance_12h import ( KernelDeniedError, PlatformKernel, ) from src.domain.app_blocks.planner import ProductIntent, select_capabilities from src.domain.app_blocks.registry import load_default_registry from src.domain.app_blocks.vertical_recipes import match_vertical_recipe from src.domain.starter_library.catalog import known_contract_ids from tests.test_phase11_pack_migration_materialization import _captured_composition ROOT = Path(__file__).resolve().parents[1] RECEIPT = ROOT / "src/domain/app_blocks/recipes/reporting_freeze.receipt.txt" AUDIT = ROOT / "docs/audits/ARCHITECTURE_LANE_REPORTING.md" REGISTRY = ROOT / "src/domain/app_blocks/platform_coverage_registry.v1.yaml" GATED = ( "payments.checkout", "membership.portal", ) STARTER_SLOTS = { "list": "record.table", "detail": "record.detail", "create": "record.create", "edit": "record.edit", "empty": "state.empty", } def reporting_composition(): return _captured_composition( "alembic/versions/150_reporting_dashboards_pack.py", "reporting-dashboards" ) def _ids(intent: ProductIntent) -> set[str]: return {block_id for _cap, block_id, _reason in select_capabilities(intent)} def test_reporting_runtime_is_canonical_without_new_kernel(): registry = load_default_registry() block = registry.require("reporting.dashboard") canonical = registry.canonical.require("reporting.dashboard") assert block.status == "implemented" assert "reporting.dashboard" in CANONICAL_BLOCK_IDS assert "reporting.dashboard" not in FEATURE_PACK_ALIASES assert "reporting.dashboard" not in feature_pack_only_aliases() assert canonical.installation.feature_packs == ("reporting-dashboards",) primitives = {dep.id for dep in canonical.requires if dep.kind == "primitive"} assert primitives == {"data.entity"} assert "payment-checkout" not in canonical.installation.feature_packs assert "saas-membership" not in canonical.installation.feature_packs assert "reporting-dashboards" in default_known_packs() composition = reporting_composition() entity = composition["entities"][0] assert entity["entity_key"] == "report_definition" keys = {field["key"] for field in entity["schema"]["fields"]} assert keys == { "name", "target_entity", "group_by_field", "measure_field", "measure_function", "status", } status = next(field for field in entity["schema"]["fields"] if field["key"] == "status") assert status["options"] == ["draft", "published"] assert "warehouse_table" not in keys assert "bi_connection_id" not in keys def test_reporting_install_plan_does_not_bind_bi_warehouse(monkeypatch): from src.domain.app_blocks import compiler registry = load_default_registry() monkeypatch.setattr( compiler, "select_capabilities", lambda _intent: [("reporting_dashboard", "reporting.dashboard", "Selected.")], ) plan = compiler.compile_install_plan(None, registry=registry) assert_install_plan_installable(plan) assert "reporting-dashboards" in plan.required_feature_packs assert "payment-checkout" not in plan.required_feature_packs assert "saas-membership" not in plan.required_feature_packs assert "data.entity" in plan.required_primitives def test_reporting_does_not_imply_billing_or_crm(): ids = _ids( ProductIntent( purpose="Staff need a reporting dashboard. No billing, orders, or customers.", source_prompt="Staff need a reporting dashboard. No billing, orders, or customers.", staff_review=True, ) ) assert "reporting.dashboard" in ids assert "billing.invoice" not in ids assert "orders.order" not in ids assert "crm.customer" not in ids assert not (ids & set(GATED)) def test_billing_and_reporting_compose_without_merging_stores(): prompt = "Staff manage invoices and want a reporting dashboard over them." ids = _ids(ProductIntent(purpose=prompt, source_prompt=prompt, staff_review=True)) assert {"billing.invoice", "reporting.dashboard"} <= ids assert not (ids & set(GATED)) def test_billing_does_not_imply_reporting(): prompt = "Staff manage invoices. No reports or dashboards." ids = _ids(ProductIntent(purpose=prompt, source_prompt=prompt, staff_review=True)) assert "billing.invoice" in ids assert "reporting.dashboard" not in ids def test_false_positives_do_not_select_reporting(): for purpose in ( "customers pay us later", "we sell services", "pricing page", "Staff manage a product catalog", "Staff manage customer orders", "We use Tableau for that already", "Power BI handles our analytics", ): ids = _ids(ProductIntent(purpose=purpose, source_prompt=purpose, staff_review=True)) assert "reporting.dashboard" not in ids, purpose def test_negative_intent_skips_reporting(): for prompt in ( "Client records for the firm. Do not add reporting.", "Staff CRM. No dashboards.", "Internal tool. No saved reports yet.", ): ids = _ids(ProductIntent(purpose=prompt, source_prompt=prompt, staff_review=True)) assert "reporting.dashboard" not in ids, prompt def test_positive_intent_selects_reporting(): for prompt in ( "Build a saved report dashboard for our staff.", "Staff need a reporting dashboard.", "Build a report definition summarizing customer invoices.", "Our team needs an aggregate report over case records.", ): ids = _ids(ProductIntent(purpose=prompt, source_prompt=prompt, staff_review=True)) assert "reporting.dashboard" in ids, prompt assert not (ids & set(GATED)), prompt def test_cross_archetype_reporting_is_the_same_block(): prompts = ( "Law firm staff want a reporting dashboard over case records.", "Repair shop wants a summary report over service jobs.", "Consulting practice wants a saved report over invoices.", ) for prompt in prompts: ids = _ids(ProductIntent(purpose=prompt, source_prompt=prompt, staff_review=True)) assert "reporting.dashboard" in ids, prompt assert "payments.checkout" not in ids def test_13a_do_not_create_orders_still_misses_reporting(): prompt = ( "Our sales office maintains catalog items in a shared list. " "Employees should sign in, add items and correct descriptions. Do not create orders." ) assert match_vertical_recipe(prompt) is None ids = _ids(ProductIntent(purpose=prompt, source_prompt=prompt, staff_review=True)) assert "catalog.product" in ids assert "reporting.dashboard" not in ids def test_reporting_reuses_record_starter_contracts(): registry = load_default_registry() block = registry.require("reporting.dashboard") contracts = { slot.id: slot.component_contract for slot in block.slots if slot.component_contract } assert STARTER_SLOTS.items() <= contracts.items() assert "report.dashboard" not in known_contract_ids() assert "stat.card" not in {contracts[k] for k in STARTER_SLOTS} def test_reporting_runtime_crud_is_staff_managed_records(): kernel = PlatformKernel(namespace="reporting-freeze") kernel.as_anonymous() with pytest.raises(KernelDeniedError): kernel.create_record("report_definition", {"name": "Weekly summary"}) kernel.as_staff() created = kernel.create_record( "report_definition", { "name": "Open invoices by status", "target_entity": "invoice_record", "group_by_field": "status", "measure_function": "count", "status": "draft", }, ) listed = kernel.list_records("report_definition") assert listed[0]["id"] == created["id"] kernel.patch_record("report_definition", created["id"], {"status": "published"}) read = kernel.get_record("report_definition", created["id"]) assert read["status"] == "published" assert read["target_entity"] == "invoice_record" other = PlatformKernel(namespace="reporting-other") other.as_staff() assert other.list_records("report_definition") == [] def test_reporting_and_billing_records_coexist_in_kernel(): kernel = PlatformKernel(namespace="reporting-billing") kernel.as_staff() invoice = kernel.create_record( "invoice_record", {"reference": "INV-1", "customer_name": "Ada", "status": "draft"}, ) report = kernel.create_record( "report_definition", {"name": "Invoices", "target_entity": "invoice_record", "measure_function": "count", "status": "draft"}, ) assert invoice["id"] != report["id"] assert kernel.get_record("invoice_record", invoice["id"])["reference"] == "INV-1" assert kernel.get_record("report_definition", report["id"])["name"] == "Invoices" def test_materialized_report_hosts_use_records_not_bi_warehouse(): import json from src.domain.app_blocks.hosts import merge_host_map 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())) reports = {h["slot"]: h for h in hosts["hosts"] if h["block"] == "reporting.dashboard"} assert {"list", "detail", "create", "edit", "empty"} <= set(reports) for slot in ("list", "detail", "create", "edit"): path = reports[slot]["implementation"].partition("#")[0] source = (root / path).read_text(encoding="utf-8") assert "report_definition" in source assert "bi_connection_id" not in source assert "/api/reporting" not in source def test_reporting_freeze_receipts_and_coverage_next_family(): receipt = RECEIPT.read_text(encoding="utf-8") audit = AUDIT.read_text(encoding="utf-8") for required in ( "reporting_admitted=yes", "canonical_app_block=reporting.dashboard", "canonical_feature_pack=reporting-dashboards", "reporting_without_bi_warehouse=yes", "family_frozen=yes", ): assert required in receipt, required assert "reporting_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 by_id["reporting.dashboard"]["status"] == "frozen" assert by_id["reporting"]["status"] == "frozen" assert by_id["reporting-dashboards"]["status"] == "frozen" assert registry["family_just_frozen"] == "visualization_calculation" def test_reporting_production_smoke_script(): from scripts.architecture_lane_reporting_smoke import run_smoke payload = run_smoke() assert payload["pass"] is True, payload["failures"] assert payload["on_disk_reporting_recipes"] == []