"""Regression guard for the P1 install-materialization defect found while building 11C1: stamp_composition() (which adds schema.workflow, schema.domain_events, field_policy flags, and schema_mode) has zero production call sites outside two one-time restamp migrations (134, 135). install_pack() reads feature_packs.composition directly from the DB with no runtime stamping step. Migrations 136-138 originally seeded RAW, unstamped composition blobs -- meaning assignment-tracking, task-management, and activity-timeline would have installed onto real projects missing their workflow graphs, domain events, and protected-schema flags entirely. Fixed by calling stamp_composition() inside each migration's own upgrade() at insert time (140 for notifications-in-app was written this way from the start). This test runs each Phase 11 pack migration's real upgrade() function against a captured (not executed) INSERT, so it exercises the exact code path that will run against a real database -- not just stamp_composition() in isolation (already covered per-pack in test_feature_pack_contract.py / test_task_management_11b2.py / test_activity_timeline_11b3.py / test_notifications_in_app_11c1.py). This is intended to become the standard proof for every future Phase 11 pack migration, not a one-off defect check. """ from __future__ import annotations import importlib.util from types import ModuleType from unittest.mock import patch import pytest import sqlalchemy as sa from unittest.mock import MagicMock def _load_migration(path: str) -> ModuleType: spec = importlib.util.spec_from_file_location(path.rsplit("/", 1)[-1][:-3], path) module = importlib.util.module_from_spec(spec) spec.loader.exec_module(module) return module def _captured_composition(migration_path: str, slug: str) -> dict: """Run the migration's real upgrade(), capturing the composition it would have inserted instead of executing SQL against a real database.""" module = _load_migration(migration_path) captured: list[dict] = [] def fake_bulk_insert(table, rows): captured.extend(rows) connection = MagicMock() def capture_execute(statement): if isinstance(statement, sa.sql.dml.Insert): captured.append(statement.compile().params) connection.execute.side_effect = capture_execute with patch("alembic.op.bulk_insert", side_effect=fake_bulk_insert), patch("alembic.op.get_bind", return_value=connection): module.upgrade() matches = [row for row in captured if row.get("slug") == slug] assert len(matches) == 1, f"expected exactly one row for {slug}, got {len(matches)}" return matches[0]["composition"] @pytest.mark.parametrize( "migration_path,slug,expects_workflow,expects_events,expected_schema_mode", [ ( "alembic/versions/136_assignment_tracking_pack.py", "assignment-tracking", True, True, "extensible", ), ( "alembic/versions/137_task_management_pack.py", "task-management", True, True, "extensible", ), ( "alembic/versions/138_activity_timeline_pack.py", "activity-timeline", False, False, "protected", ), ( "alembic/versions/140_notifications_in_app_pack.py", "notifications-in-app", False, # 11C2: owns notification.email_delivery_attempted (a fact about # its own email_status field, not a business event impersonation # -- see test_notifications_email_11c2.py). True, "protected", ), ( "alembic/versions/141_file_attachments_pack.py", "file-attachments", False, False, "protected", ), ( "alembic/versions/142_documents_lifecycle_pack.py", "documents-lifecycle", False, False, "extensible", ), ( "alembic/versions/144_customer_records_pack.py", "customer-records", False, False, "extensible", ), ( "alembic/versions/146_order_management_pack.py", "order-management", False, False, "extensible", ), ( "alembic/versions/147_billing_management_pack.py", "billing-management", False, False, "extensible", ), ( "alembic/versions/148_inventory_management_pack.py", "inventory-management", False, False, "extensible", ), ( "alembic/versions/149_membership_management_pack.py", "membership-management", False, False, "extensible", ), ], ) def test_migration_persists_fully_materialized_composition( migration_path, slug, expects_workflow, expects_events, expected_schema_mode ): """persisted pack composition == materialized composition: what the migration actually inserts must already be the fully stamped shape (schema.workflow / schema.domain_events / schema_mode present exactly where the pack's design calls for them), not a raw blob depending on a stamping step that never runs again after insert.""" composition = _captured_composition(migration_path, slug) entity_schema = composition["entities"][0]["schema"] assert composition["schema_version"] == 2, "must be the stamped v2 contract, not a raw v1 blob" assert ("workflow" in entity_schema) is expects_workflow assert ("domain_events" in entity_schema) is expects_events assert entity_schema["schema_mode"] == expected_schema_mode assert composition["provides"], "every pack must provide at least one capability id" def test_assignment_created_recipient_route_survives_migration_materialization(): """11R4: NotificationRoute is Core contract data added to the Python model, not automatically transported. Prove it actually reaches the same schema.domain_events payload Cloud reads -- the exact defect class found in the 11B1-11B3 migrations (raw composition persisted instead of the fully materialized shape) could silently drop `recipient` the same way it silently dropped `schema.workflow` / `schema.domain_events` themselves.""" composition = _captured_composition( "alembic/versions/136_assignment_tracking_pack.py", "assignment-tracking" ) assignment_entity = next( entity for entity in composition["entities"] if entity["entity_key"] == "assignment" ) domain_events = assignment_entity["schema"]["domain_events"] created = next(evt for evt in domain_events if evt["id"] == "assignment.created") assert created["recipient"] == { "strategy": "assignment_assignee", "exclude_actor": False, } status_changed = next( evt for evt in domain_events if evt["id"] == "assignment.status_changed" ) assert "recipient" not in status_changed, ( "11R4 scope is assignment.created only -- routing must not leak onto " "sibling events that never declared a recipient" ) def _stamped_approval_workflow_entity() -> dict: """approval-workflow was seeded at migration 103, well before stamp_composition() existed -- unlike 136-140 it never calls stamp_composition() in its own upgrade(). Its composition was instead swept up, once, by migration 134's historical backfill (SELECT every feature_packs row, stamp it, UPDATE in place -- see 134's upgrade()). That backfill already ran; verifying "the current DB row is correct" today means proving stamp_composition() applied to 103's exact raw composition produces the same materialized shape 134 would have written, which is what this helper does instead of re-deriving it from a synthetic fixture (test_feature_pack_contract.py's is intentionally minimal).""" from unittest.mock import patch from src.domain.feature_packs.normalize import stamp_composition module = _load_migration("alembic/versions/103_feature_packs_seed_3.py") captured: list[dict] = [] def fake_bulk_insert(table, rows): captured.extend(rows) with patch("alembic.op.bulk_insert", side_effect=fake_bulk_insert): module.upgrade() raw = next(row for row in captured if row.get("slug") == "approval-workflow")["composition"] stamped = stamp_composition("approval-workflow", raw) return next(entity for entity in stamped["entities"] if entity["entity_key"] == "approval_request") def test_approval_requested_role_members_route_survives_historical_backfill_materialization(): """11R4B: role_members on approval.requested survives the real historical backfill path (Sec docstring on _stamped_approval_workflow_entity).""" domain_events = _stamped_approval_workflow_entity()["schema"]["domain_events"] requested = next(evt for evt in domain_events if evt["id"] == "approval.requested") assert requested["recipient"] == { "strategy": "role_members", "role_field": "approver_role", "exclude_actor": False, } def test_approval_outcome_events_field_user_ref_route_survives_historical_backfill_materialization(): """11R4C: field_user_ref on all three approval outcome events survives the same historical backfill path, and role_members stays confined to approval.requested -- neither route leaks onto the other's events.""" domain_events = _stamped_approval_workflow_entity()["schema"]["domain_events"] requested = next(evt for evt in domain_events if evt["id"] == "approval.requested") assert requested["recipient"]["strategy"] == "role_members", ( "approval.requested must keep its own role_members route, not be " "overwritten by the field_user_ref work done for its sibling events" ) for outcome_id in ("approval.approved", "approval.rejected", "approval.escalated"): outcome = next(evt for evt in domain_events if evt["id"] == outcome_id) assert outcome["recipient"] == { "strategy": "field_user_ref", "field": "requested_by", "exclude_actor": False, }, f"'{outcome_id}' must materialize field_user_ref/requested_by" def test_notifications_migration_also_persists_hand_embedded_access_policy(): """notifications-in-app's access_policy isn't produced by stamp_composition at all (confirmed: it discards its access_policy parameter) -- it must survive from the migration's own raw entity definition into the exact payload that would be persisted.""" composition = _captured_composition( "alembic/versions/140_notifications_in_app_pack.py", "notifications-in-app" ) policy = composition["entities"][0]["schema"]["access_policy"] assert policy["version"] == 1 assert "create" not in policy["roles"]["member"] assert policy["roles"]["member"]["read"]["scope"] == "owner" def test_documents_migration_also_persists_hand_embedded_access_policy(): """documents-lifecycle (11D5B) follows the same hand-embedded-access_policy pattern as notifications-in-app/file-attachments, for the same reason: a member's documents are per-user data, and catalog_v2's `security` block is never translated into schema.access_policy (confirmed in normalize.py -- stamp_composition discards its access_policy parameter). Unlike notifications (read/update only, no member create), a member has full CRUD at scope=owner here -- documents are member-authored content, and 11D5C's runtime CRUD proof depends on this access_policy already being correct.""" composition = _captured_composition( "alembic/versions/142_documents_lifecycle_pack.py", "documents-lifecycle" ) policy = composition["entities"][0]["schema"]["access_policy"] assert policy["version"] == 1 for action in ("create", "read", "update", "delete"): assert policy["roles"]["member"][action]["scope"] == "owner" assert policy["roles"]["admin"][action]["scope"] == "all" assert policy["roles"]["super_admin"][action]["scope"] == "all" def test_documents_entity_content_format_admits_only_plain_text(): """11D5A froze content_format=plain_text for 11D5, with markdown/rich_text explicitly deferred to a future capability slice (its own editor + sanitizer + render-boundary work). This must not silently widen.""" composition = _captured_composition( "alembic/versions/142_documents_lifecycle_pack.py", "documents-lifecycle" ) fields = composition["entities"][0]["schema"]["fields"] content_format = next(f for f in fields if f["key"] == "content_format") assert content_format["options"] == ["plain_text"] assert "workflow" not in composition["entities"][0]["schema"], ( "status must stay a plain select field -- workflow.transition was " "explicitly rejected for document status in the 11D5A audit" ) def test_documents_entity_sets_rls_enabled_for_create_path_owner_stamping(): """11D5C load-bearing finding: resolveCreateOwner (pc-gateway records.ts) gates owner stamping on entity.rls_enabled/owner_managed ONLY -- unlike resolveRecordScope, it does not consult schema.access_policy at all. Without this top-level flag, every member-created document would get owner_user_id=null and scope=owner reads would match nothing. This is the entity-level flag, not inside schema -- must survive materialization as a sibling of "schema", not nested inside it.""" composition = _captured_composition( "alembic/versions/142_documents_lifecycle_pack.py", "documents-lifecycle" ) assert composition["entities"][0]["rls_enabled"] is True def test_documents_content_field_max_bytes_survives_migration_materialization(): """11D5C: document.content must carry the 256 KiB (262144 byte) ceiling into the exact payload that would be persisted -- this is the field Cloud's validateRecordData reads to stop content from becoming an alternate blob-storage path. A regression here would silently remove the size guard without any test noticing.""" composition = _captured_composition( "alembic/versions/142_documents_lifecycle_pack.py", "documents-lifecycle" ) fields = composition["entities"][0]["schema"]["fields"] content = next(f for f in fields if f["key"] == "content") assert content["max_bytes"] == 262144 title = next(f for f in fields if f["key"] == "title") assert "max_bytes" not in title, ( "max_bytes is opt-in per field -- title was deliberately left " "unbounded in 11D5C (out of scope, see the 11D5C audit's Sec 10)" )