"""Phase 11D3: files.attachments composition + provider-neutrality proofs.""" from __future__ import annotations import json from pathlib import Path from src.domain.app_blocks import compiler from src.domain.app_blocks.hosts import host_map_payload from src.domain.app_blocks.planner import ProductIntent, default_slots, select_capabilities from src.domain.app_blocks.public import _BLOCK_COPY from src.domain.app_blocks.registry import load_default_registry from src.domain.feature_packs.normalize import stamp_composition _MANIFEST = Path(__file__).resolve().parents[1] / "src/domain/app_blocks/manifests/files.attachments.v1.yaml" _PROVIDER_LEAKS = ( "r2", "cloudflare", "amazon ses", "ses sdk", "s3-compatible", "storage_key", "object_key", "bucket_name", "access_key", "secret_access_key", ) def test_planner_matches_tight_attachment_phrases_only(): hit = select_capabilities(ProductIntent(purpose="Attach files to cases and leads")) assert any(block_id == "files.attachments" for _cap, block_id, _reason in hit) miss = select_capabilities(ProductIntent(purpose="Users can upload a profile photo later")) assert not any(block_id == "files.attachments" for _cap, block_id, _reason in miss) def test_host_catalog_lists_attachment_slots(): payload = host_map_payload() keys = {(item["block"], item["slot"]) for item in payload["hosts"]} assert ("files.attachments", "pipeline") in keys assert ("files.attachments", "detail") in keys assert ("files.attachments", "empty") in keys assert ("files.attachments", "upload") in keys def test_materialized_attachment_hosts_use_generic_records_not_a_private_store(): from src.domain.app_blocks.hosts import merge_host_map root = Path(__file__).resolve().parents[1] for framework in ("html", "nuxt_app", "react_app"): folder = root / "boilerplates" / framework hosts = merge_host_map(json.loads((folder / "p2hs/app-block-hosts.json").read_text())) mapped = {h["slot"]: h for h in hosts["hosts"] if h["block"] == "files.attachments"} assert {"pipeline", "detail", "empty"} <= set(mapped) for slot in ("pipeline", "detail"): assert mapped[slot]["implementation"] path = mapped[slot]["implementation"].partition("#")[0] source = (folder / path).read_text(encoding="utf-8") assert "file_attachment" in source assert "storage_key" not in source assert "object_key" not in source assert "bucket" not in source assert mapped["empty"]["implementation"] def test_public_label_exists_and_does_not_leak_the_raw_id(): label = _BLOCK_COPY["files.attachments"] assert label == "File attachments" assert label != "files.attachments" def test_install_plan_resolves_file_attachments_pack_and_stays_satisfiable(monkeypatch): registry = load_default_registry() monkeypatch.setattr( compiler, "select_capabilities", lambda _intent: [("files_attachments", "files.attachments", "Selected.")], ) plan = compiler.compile_install_plan(None, registry=registry) assert not plan.unsatisfiable assert plan.required_feature_packs == ["file-attachments"] assert "data.entity" in plan.required_primitives block_ids = {b.id for b in plan.blocks} expected = {"files.attachments", "admin.shell", "admin.record_table", "admin.record_detail"} assert expected <= block_ids def test_missing_file_attachments_pack_is_an_honest_gap_not_a_fallback(monkeypatch): from src.domain.app_blocks.catalogs import default_known_packs registry = load_default_registry() monkeypatch.setattr( compiler, "select_capabilities", lambda _intent: [("files_attachments", "files.attachments", "Selected.")], ) plan = compiler.compile_install_plan( None, registry=registry, known_packs=default_known_packs() - {"file-attachments"} ) assert plan.unsatisfiable assert any( g.type == "feature_pack_missing" and g.id == "file-attachments" and g.severity == "build_blocking" for g in plan.gaps ) def test_attachment_entity_is_subject_agnostic_and_file_id_only(): stamped = stamp_composition( "file-attachments", { "requires": [], "entities": [ { "entity_key": "file_attachment", "schema": { "fields": [ {"key": "file_id", "type": "text", "required": True, "indexed": True}, { "key": "subject_entity_key", "type": "text", "required": True, "indexed": True, }, { "key": "subject_record_id", "type": "text", "required": True, "indexed": True, }, ] }, } ], }, ) entity = stamped["entities"][0] keys = [field["key"] for field in entity["schema"]["fields"]] assert keys == ["file_id", "subject_entity_key", "subject_record_id"] assert entity["schema"]["schema_mode"] == "protected" assert "workflow" not in entity["schema"] assert "domain_events" not in entity["schema"] for forbidden in ("storage_key", "object_key", "bucket", "lead_file_id", "case_file_id"): assert forbidden not in keys def test_same_contract_covers_task_case_and_lead_subjects(): subjects = ("platform_task", "case_record", "lead") for subject in subjects: stamped = stamp_composition( "file-attachments", { "requires": [], "entities": [ { "entity_key": "file_attachment", "schema": { "fields": [ {"key": "file_id", "type": "text"}, {"key": "subject_entity_key", "type": "text"}, {"key": "subject_record_id", "type": "text"}, ] }, } ], }, ) keys = {field["key"] for field in stamped["entities"][0]["schema"]["fields"]} assert "subject_entity_key" in keys assert f"{subject}_file_id" not in keys def test_product_contracts_do_not_expose_storage_or_email_providers(): from src.domain.feature_packs.normalize import load_pack_contract contract = load_pack_contract("file-attachments", {"requires": [], "entities": []}) registry = load_default_registry() manifest = registry.require("files.attachments") hosts = [ item for item in host_map_payload()["hosts"] if item["block"] == "files.attachments" ] surfaces = [ _MANIFEST.read_text(encoding="utf-8"), manifest.description, " ".join(manifest.tags), _BLOCK_COPY["files.attachments"], " ".join(contract.provide_ids()), " ".join(contract.required_primitive_ids()), str(default_slots(manifest, ProductIntent())), str(hosts), ] blob = "\n".join(surfaces).lower() for leak in _PROVIDER_LEAKS: assert leak not in blob, f"product contract leaked {leak!r}" assert contract.configuration_providers() == [] assert contract.provider_bindings == []