"""Phase 11C1: notifications.in_app end-to-end proof. Governing rule (same class as 11B3, applied to a second, independent consumer): in-app notifications consume canonical domain events and create user-facing notification projections. They do not emit business events on behalf of the originating capability and do not become a second event bus. Inspect-phase finding: the platform already has a deterministic individual- user identity -- the authenticated app_user's JWT `sub` claim, exposed as CallerAuth.userId in P2HSCloud's apps/pc-gateway/src/lib/auth-context.ts, and already used by resolveRecordScope()/checkAccessPolicy() to scope "owner" records. No new relation type or identity primitive was needed; a notification record's recipient is expressed as the SAME owner mechanism every existing rls_enabled/access_policy entity already relies on. Read/unread is deliberately NOT workflow.transition: read_at is a plain, nullable timestamp field (application state), not a governed lifecycle -- see docs/audits/NOTIFICATIONS_IN_APP_11C1.md Sec 2 for why. """ from __future__ import annotations 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, 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.catalog_v2 import CATALOG_OVERLAYS from src.domain.feature_packs.normalize import load_pack_contract, stamp_composition def test_planner_matches_tight_in_app_notification_phrases_only(): hit = select_capabilities(ProductIntent(purpose="We need an in-app notification inbox")) assert any(block_id == "notifications.in_app" for _cap, block_id, _reason in hit) # Bare "notify"/"notification" is far too common to match as a bag of words. miss = select_capabilities(ProductIntent(purpose="Notify staff when a form is submitted")) assert not any(block_id == "notifications.in_app" for _cap, block_id, _reason in miss) def test_host_catalog_lists_notification_slots(): payload = host_map_payload() keys = {(item["block"], item["slot"]) for item in payload["hosts"]} assert ("notifications.in_app", "pipeline") in keys assert ("notifications.in_app", "detail") in keys assert ("notifications.in_app", "empty") in keys def test_public_label_exists_and_does_not_leak_the_raw_id(): label = _BLOCK_COPY["notifications.in_app"] assert label == "In-app notifications" assert label != "notifications.in_app" def test_install_plan_resolves_notifications_pack_and_stays_satisfiable(monkeypatch): registry = load_default_registry() monkeypatch.setattr( compiler, "select_capabilities", lambda _intent: [("notifications_in_app", "notifications.in_app", "Selected.")], ) plan = compiler.compile_install_plan(None, registry=registry) assert not plan.unsatisfiable assert plan.required_feature_packs == ["notifications-in-app"] # provider_dependency=no: in-app notifications need no provider binding at all. assert "workflow.transition" not in plan.required_primitives block_ids = {b.id for b in plan.blocks} expected = {"notifications.in_app", "admin.shell", "admin.record_table", "admin.record_detail"} assert expected <= block_ids def test_missing_notifications_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: [("notifications_in_app", "notifications.in_app", "Selected.")], ) plan = compiler.compile_install_plan( None, registry=registry, known_packs=default_known_packs() - {"notifications-in-app"} ) assert plan.unsatisfiable assert any( g.type == "feature_pack_missing" and g.id == "notifications-in-app" and g.severity == "build_blocking" for g in plan.gaps ) def test_notifications_cannot_invent_a_new_canonical_event_identity(): """Required failure mode #1, refined for 11C2: this pack may own an event about its OWN delivery-channel state (notification.email_delivery_attempted -- a field.changed fact about its own entity, exactly like contact-capture already has contact.notification_sent/_failed), but it must never emit, alias, or otherwise fabricate an event that could be confused with an ORIGINATING capability's business event (assignment.created, task.*, ...). It must also still require zero new primitives -- email.send is reused, not invented -- and zero provider bindings (see catalog_v2.py's comment on why declaring one here would exercise an unproven code path).""" contract = load_pack_contract( "notifications-in-app", {"requires": [], "entities": []}, ) assert [e.id for e in contract.events] == ["notification.email_delivery_attempted"] event = contract.events[0] assert event.producer.kind == "field.changed" assert event.producer.entity == "platform_notification" assert event.aliases == [] # Never impersonates an originating capability's namespace. assert not event.id.startswith(("assignment.", "task.", "lead.", "case.", "contact.")) assert contract.required_primitive_ids() == [] assert contract.provider_bindings == [] def test_notifications_cannot_be_written_directly_by_generated_app_code(): """Required failure mode #3 (checked before #2 since it grounds it): no public write path, no Codegen-facing helper, protected schema_mode.""" contract = load_pack_contract( "notifications-in-app", {"requires": [], "entities": []}, ) assert "form.public_create" not in contract.required_primitive_ids() assert contract.security.public_actions == [] raw = CATALOG_OVERLAYS["notifications-in-app"] assert raw.get("function_slugs") in (None, []) entity_meta = next(e for e in raw["entities"] if e.entity_key == "platform_notification") assert entity_meta.schema_mode == "protected" def test_notifications_cannot_bypass_recipient_authorization(): """Required failure mode #2: the hand-embedded access_policy (see 140_notifications_in_app_pack.py) must, if it were pushed to Cloud, match exactly what checkAccessPolicy() (packages/shared/src/records.ts) expects: no role granted "create" at all (only a non-app_user/internal caller may create a notification -- recipient_identity_deterministic + direct_codegen_notification_write=no), and every "member" grant scoped to "owner", never "all" -- one recipient can never read another's rows.""" import importlib.util spec = importlib.util.spec_from_file_location( "m140", "alembic/versions/140_notifications_in_app_pack.py" ) module = importlib.util.module_from_spec(spec) spec.loader.exec_module(module) policy = module._NOTIFICATION_ACCESS_POLICY assert policy["version"] == 1 for role, actions in policy["roles"].items(): assert "create" not in actions, f"{role} must never be granted create" for action, entry in actions.items(): assert entry["effect"] == "allow" if role == "member": assert entry["scope"] == "owner", ( f"member's {action} must be owner-scoped, never all-scoped" ) else: assert entry["scope"] == "all" # Confirm stamp_composition (which every other field-flag pass runs # through) does not strip or alter this hand-embedded policy. raw = { "function_slugs": [], "entities": [module._PLATFORM_NOTIFICATION_ENTITY], "connector_actions": [], "requires": [], } stamped = stamp_composition("notifications-in-app", raw) assert stamped["entities"][0]["schema"]["access_policy"] == policy def test_read_at_is_not_a_workflow_field(): """Read state is application state, not a business lifecycle -- see docs/audits/NOTIFICATIONS_IN_APP_11C1.md Sec 2.""" contract = load_pack_contract("notifications-in-app", {}) assert contract.workflows == [] stamped = stamp_composition( "notifications-in-app", { "requires": [], "entities": [ { "entity_key": "platform_notification", # email_status also present -- 11C2's field.changed event # producer validates its field exists on this entity. "schema": { "fields": [ {"key": "read_at", "type": "text"}, { "key": "email_status", "type": "select", "options": [ "not_requested", "sent", "failed", "configuration_required", ], }, ] }, } ], }, ) assert "workflow" not in stamped["entities"][0]["schema"] def _project_envelope(canonical_event_id: str, producer, entity: str, target: str) -> dict: """Same proof-harness style as 11B3's activity mapping, applied to a second, independent consumer -- proves multiple consumers share one event authority without coordinating with each other.""" if target == "activity": return { "canonical_event_id": canonical_event_id, "subject_entity_key": entity, "subject_record_id": "", "actor_type": "service", "actor_id": "", "occurred_at": "", "producer_capability": "", "payload": {}, } return { "recipient_user_id": "", "source_event_id": "", "canonical_event_id": canonical_event_id, "subject_entity_key": entity, "subject_record_id": "", "title": "", "producer_capability": "", "payload": {}, "read_at": None, } def test_same_canonical_event_feeds_activity_timeline_and_notifications_independently(): """The key multi-consumer proof: one real canonical event (assignment. status_changed) projects cleanly through BOTH activity.timeline's and notifications.in_app's independent field shapes, sharing only the event's canonical id and subject reference -- proving one event authority serves multiple consumers without either needing to know about the other.""" contract = load_pack_contract("assignment-tracking", {}) event = next(e for e in contract.events if e.id == "assignment.status_changed") assert event.producer is not None activity_row = _project_envelope(event.id, event.producer, event.producer.entity, "activity") notification_row = _project_envelope( event.id, event.producer, event.producer.entity, "notification" ) # Same source fact, two independently-shaped projections. assert activity_row["canonical_event_id"] == notification_row["canonical_event_id"] == event.id assert ( activity_row["subject_entity_key"] == notification_row["subject_entity_key"] == event.producer.entity ) # Neither projection's field set is a subset/superset hack of the other -- # each is shaped for its own consumer (activity has no recipient/read # state; notifications has no actor_type/occurred_at). assert "recipient_user_id" not in activity_row assert "actor_type" not in notification_row