"""Wave 2: quote_line, invoice_line, catalog_variant as related records. Category is an indexed text field on catalog_item. Header amount/price stay explicit parent fields. No new Cloud jobs or quoting/SKU engines. """ from __future__ import annotations import json import re from pathlib import Path import pytest import yaml from src.domain.app_blocks.generated_product_conformance_12h import ( ENTITY_BY_BLOCK, KernelDeniedError, PlatformKernel, ) from src.domain.app_blocks.hosts import merge_host_map from src.domain.app_blocks.registry import load_default_registry from src.domain.feature_packs.catalog_v2 import overlay_for from src.integrations.platform_ports import QUEUE_REQUEST_TYPES from tests.test_phase11_pack_migration_materialization import ( _captured_composition, _load_migration, ) ROOT = Path(__file__).resolve().parents[1] REGISTRY = ROOT / "src/domain/app_blocks/platform_coverage_registry.v1.yaml" WAVE2 = "alembic/versions/158_quote_invoice_catalog_child_records.py" CHILD_SLOTS = { "crm.quote": ("lines", "line_create", "line_detail"), "billing.invoice": ("lines", "line_create", "line_detail"), "catalog.product": ("variants", "variant_create", "variant_detail"), } CHILD_ENTITIES = { "crm.quote": "quote_line", "billing.invoice": "invoice_line", "catalog.product": "catalog_variant", } PARENT_FIELDS = { "quote_line": "quote_record_id", "invoice_line": "invoice_record_id", "catalog_variant": "catalog_item_record_id", } def _live(slug: str) -> dict: module = _load_migration(WAVE2) return { "quote-lifecycle": module.quote_lifecycle_v1_1, "billing-management": module.billing_management_v1_1, "product-catalog": module.product_catalog_v1_1, }[slug]() def _entity(composition: dict, key: str) -> dict: return next(item for item in composition["entities"] if item["entity_key"] == key) def _keys(entity: dict) -> set[str]: return {field["key"] for field in entity["schema"]["fields"]} def _host_fragment(root: Path, implementation: str) -> str: rel, _, symbol = implementation.partition("#") text = (root / rel).read_text(encoding="utf-8") if not symbol: return text match = re.search( rf"(?:export\s+)?(?:async\s+)?function\s+{re.escape(symbol)}\b", text, ) if not match: return text rest = text[match.end() :] nxt = re.search(r"\nexport\s+(?:async\s+)?function\s+", rest) end = match.end() + nxt.start() if nxt else len(text) return text[match.start() : end] def test_historical_pack_migrations_keep_parent_only_schema(): quote = _captured_composition( "alembic/versions/152_quote_lifecycle_pack.py", "quote-lifecycle" ) invoice = _captured_composition( "alembic/versions/147_billing_management_pack.py", "billing-management" ) catalog = _captured_composition( "alembic/versions/145_product_catalog_pack.py", "product-catalog" ) assert [item["entity_key"] for item in quote["entities"]] == ["quote"] assert [item["entity_key"] for item in invoice["entities"]] == ["invoice_record"] assert [item["entity_key"] for item in catalog["entities"]] == ["catalog_item"] assert "category" not in _keys(catalog["entities"][0]) assert "quote_line" not in {item["entity_key"] for item in quote["entities"]} def test_live_wave2_composition_adds_related_children_without_jobs(): quote = _live("quote-lifecycle") invoice = _live("billing-management") catalog = _live("product-catalog") assert [item["entity_key"] for item in quote["entities"]] == [ "quote", "quote_line", ] assert [item["entity_key"] for item in invoice["entities"]] == [ "invoice_record", "invoice_line", ] assert [item["entity_key"] for item in catalog["entities"]] == [ "catalog_item", "catalog_variant", ] for composition in (quote, invoice, catalog): assert composition["function_slugs"] == [] assert composition["connector_actions"] == [] parent_quote = _entity(quote, "quote") parent_invoice = _entity(invoice, "invoice_record") parent_item = _entity(catalog, "catalog_item") assert "amount" in _keys(parent_quote) assert "amount" in _keys(parent_invoice) assert "price_amount" in _keys(parent_item) assert "category" in _keys(parent_item) line = _entity(quote, "quote_line") parent = next( field for field in line["schema"]["fields"] if field["key"] == "quote_record_id" ) assert parent["type"] == "relation" assert parent["relation_entity_key"] == "quote" assert parent["indexed"] is True variant = _entity(catalog, "catalog_variant") rel = next( field for field in variant["schema"]["fields"] if field["key"] == "catalog_item_record_id" ) assert rel["type"] == "relation" assert rel["relation_entity_key"] == "catalog_item" def test_overlays_declare_child_entities_on_existing_packs(): quote_keys = {entity.entity_key for entity in overlay_for("quote-lifecycle")["entities"]} invoice_keys = {entity.entity_key for entity in overlay_for("billing-management")["entities"]} catalog_keys = {entity.entity_key for entity in overlay_for("product-catalog")["entities"]} assert quote_keys == {"quote", "quote_line"} assert invoice_keys == {"invoice_record", "invoice_line"} assert catalog_keys == {"catalog_item", "catalog_variant"} def test_app_block_slots_are_record_contracts_not_job_ports(): registry = load_default_registry() for block_id, slots in CHILD_SLOTS.items(): block = registry.require(block_id) by_id = {slot.id: slot for slot in block.slots} for slot_id in slots: slot = by_id[slot_id] assert slot.type == "ui_slot" assert slot.query_source is None assert slot.component_contract in { "record.table", "record.create", "record.detail", } requires = registry.canonical.require(block_id).requires primitives = {dep.id for dep in requires if dep.kind == "primitive"} assert primitives == {"data.entity"} assert ENTITY_BY_BLOCK["crm.quote"] == "quote" assert ENTITY_BY_BLOCK["billing.invoice"] == "invoice_record" assert ENTITY_BY_BLOCK["catalog.product"] == "catalog_item" def test_registry_drops_line_variant_and_category_gaps(): registry = yaml.safe_load(REGISTRY.read_text(encoding="utf-8")) by_id = {row["id"]: row for row in registry["capabilities"]} assert by_id["crm.quote"]["gaps"] == [] assert by_id["catalog"]["gaps"] == [] assert by_id["catalog.product"]["gaps"] == [] assert "line_items" not in by_id["billing.invoice"]["gaps"] assert "variants" not in by_id["catalog.product"]["gaps"] assert "categories" not in by_id["catalog.product"]["gaps"] def test_no_new_cloud_jobs_for_lines_or_variants(): from src.domain.cloud.service import _REQUEST_TYPE_POLICY forbidden = { "quote.line", "invoice.line", "catalog.variant", "quote.line.add", "invoice.line.add", "catalog.variant.upsert", } assert not (forbidden & set(QUEUE_REQUEST_TYPES)) assert not (forbidden & set(_REQUEST_TYPE_POLICY)) def test_header_amount_and_price_stay_explicit_after_child_writes(): kernel = PlatformKernel(namespace="wave2-children") kernel.as_anonymous() with pytest.raises(KernelDeniedError): kernel.create_record("quote_line", {"description": "Labor"}) kernel.as_staff() quote = kernel.create_record( "quote", {"reference": "Q-1", "status": "draft", "amount": 100, "currency": "USD"}, ) kernel.create_record( "quote_line", { "quote_record_id": quote["id"], "description": "Labor", "quantity": 2, "unit_price": 40, }, ) kernel.create_record( "quote_line", { "quote_record_id": quote["id"], "description": "Parts", "quantity": 1, "unit_price": 15, }, ) assert kernel.get_record("quote", quote["id"])["amount"] == 100 assert len(kernel.list_records("quote_line")) == 2 invoice = kernel.create_record( "invoice_record", {"reference": "INV-1", "status": "draft", "amount": 250, "currency": "USD"}, ) kernel.create_record( "invoice_line", { "invoice_record_id": invoice["id"], "description": "Retainer", "quantity": 1, "unit_price": 250, }, ) assert kernel.get_record("invoice_record", invoice["id"])["amount"] == 250 item = kernel.create_record( "catalog_item", { "name": "Widget", "sku": "W1", "category": "hardware", "price_amount": 12, "currency": "USD", "status": "active", }, ) kernel.create_record( "catalog_variant", { "catalog_item_record_id": item["id"], "sku": "W1-BLUE", "name": "Blue", "price_amount": 14, "currency": "USD", "status": "active", }, ) read = kernel.get_record("catalog_item", item["id"]) assert read["price_amount"] == 12 assert read["category"] == "hardware" def test_materialized_child_hosts_use_relative_records_filter(): 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())) for block_id, slots in CHILD_SLOTS.items(): rows = {item["slot"]: item for item in hosts["hosts"] if item["block"] == block_id} entity_key = CHILD_ENTITIES[block_id] parent_field = PARENT_FIELDS[entity_key] for slot_id in slots: source = _host_fragment(root, rows[slot_id]["implementation"]) assert entity_key in source assert "/api/jobs/" not in source assert "/v1/projects/" not in source assert "cloudgate" not in source assert "INTERNAL_CONTROL_TOKEN" not in source if slot_id in {"lines", "variants"}: assert "filter" in source assert parent_field in source assert "parentRecordId" in source if slot_id in {"line_create", "variant_create"}: assert parent_field in source assert "parentRecordId" in source schema = json.loads( (root / _schema_dir(framework) / f"{entity_key}.json").read_text(encoding="utf-8") ) keys = {field["key"] for field in schema["fields"]} assert parent_field in keys catalog_schema = json.loads( (root / _schema_dir(framework) / "catalog_item.json").read_text(encoding="utf-8") ) assert "category" in {field["key"] for field in catalog_schema["fields"]} def _schema_dir(framework: str) -> Path: return { "html": Path("js/entity-schemas"), "nuxt_app": Path("lib/entity-schemas"), "react_app": Path("src/lib/entity-schemas"), }[framework]