"""compose_and_register_pack: DB-dependent validation (function_slugs must exist and be published), find-or-create of the project_local FeaturePack row, and delegation to the existing, unmodified PackInstallService.install_pack. install_pack itself is already covered elsewhere (test_feature_pack_composition.py, test_project_auth_service.py) -- this file patches it (same convention as test_codegen_retry_run.py patching private helpers) rather than re-faking its entire internal DB interaction, since composer.py's own job is only to hand off to it correctly. """ from __future__ import annotations from types import SimpleNamespace from unittest.mock import AsyncMock, patch from uuid import uuid4 import pytest from src.domain.feature_packs.composer import ( ComposedEntity, PrimitiveGap, _cloud_entity_dict, compose_and_register_pack, derive_pack_composition, ) from src.domain.feature_packs.requirements_spec import CompositionRequirement def _quote_request(**overrides) -> CompositionRequirement: payload = { "name": "quote_request", "actors": ["anonymous", "admin"], "requirements": [ {"action": "create", "resource": "quote_request", "actor": "anonymous"}, {"action": "read", "resource": "quote_request", "actor": "admin"}, ], "fields": [{"key": "name", "label": "Name", "type": "text", "required": True}], } payload.update(overrides) return CompositionRequirement.model_validate(payload) def test_derive_produces_the_confirmed_access_patterns(): spec = derive_pack_composition(_quote_request(), "p78ot2o6") assert spec.slug == "local-p78ot2o6-quote_request" assert spec.entities[0].rls_enabled is False # no "member" actor -> plain entity, not RLS # anonymous never gets an access_policy entry -- routed entirely through # form.public_create instead (see composer.py's derive_pack_composition). # super_admin is always present alongside admin (inherits its actions). assert set(spec.access_policy.roles) == {"admin", "super_admin"} assert spec.access_policy.roles["admin"]["read"].scope == "all" assert "form.public_create" in spec.requires def test_derive_sets_rls_for_a_member_actor(): req = _quote_request( actors=["member", "admin"], requirements=[ {"action": "create", "resource": "quote_request", "actor": "member"}, {"action": "read", "resource": "quote_request", "actor": "member"}, {"action": "read", "resource": "quote_request", "actor": "admin"}, ], ) spec = derive_pack_composition(req, "p78ot2o6") assert spec.entities[0].rls_enabled is True assert spec.access_policy.roles["member"]["read"].scope == "owner" assert spec.access_policy.roles["admin"]["read"].scope == "all" def test_super_admin_inherits_admins_actions_even_when_never_named(): """super_admin must never end up more restricted than admin by simple omission from a requirement that only named "admin" -- composer.py bakes this union in so Cloud's checkAccessPolicy stays a dumb exact lookup with no hierarchy of its own.""" req = _quote_request( actors=["admin"], requirements=[ {"action": "read", "resource": "quote_request", "actor": "admin"}, {"action": "delete", "resource": "quote_request", "actor": "admin"}, ], ) spec = derive_pack_composition(req, "p78ot2o6") assert "super_admin" not in {"admin"} # sanity: requirement truly never named it assert set(spec.access_policy.roles["super_admin"]) == {"read", "delete"} assert spec.access_policy.roles["super_admin"]["delete"].scope == "all" def test_super_admins_own_extra_action_survives_the_admin_union(): req = _quote_request( actors=["admin", "super_admin"], requirements=[ {"action": "read", "resource": "quote_request", "actor": "admin"}, {"action": "delete", "resource": "quote_request", "actor": "super_admin"}, ], ) spec = derive_pack_composition(req, "p78ot2o6") # super_admin keeps its own explicitly-declared "delete" AND inherits admin's "read". assert set(spec.access_policy.roles["super_admin"]) == {"read", "delete"} def test_anonymous_paired_with_anything_but_create_is_invalid(): req = _quote_request( requirements=[{"action": "read", "resource": "quote_request", "actor": "anonymous"}] ) with pytest.raises(ValueError, match="anonymous actor may only be paired with action=create"): derive_pack_composition(req, "p78ot2o6") def _ticket(**overrides) -> CompositionRequirement: payload = { "name": "support_ticket", "actors": ["member", "admin"], "requirements": [ {"action": "create", "resource": "support_ticket", "actor": "member"}, {"action": "update", "resource": "support_ticket", "actor": "admin"}, ], "fields": [ {"key": "subject", "label": "Subject", "type": "text", "required": True}, {"key": "status", "label": "Status", "type": "select", "options": ["open", "pending", "resolved"]}, ], "workflow": { "field": "status", "transitions": {"open": ["pending"], "pending": ["open", "resolved"], "resolved": []}, }, } payload.update(overrides) return CompositionRequirement.model_validate(payload) def test_workflow_requirement_composes_with_the_transition_primitive(): spec = derive_pack_composition(_ticket(), "p78ot2o6") assert not isinstance(spec, PrimitiveGap) assert "workflow.transition" in spec.requires assert spec.entities[0].workflow is not None assert spec.entities[0].workflow.field == "status" assert spec.entities[0].workflow.transitions["pending"] == ["open", "resolved"] def test_workflow_field_not_declared_among_fields_is_a_hard_failure(): req = _ticket(workflow={"field": "does_not_exist", "transitions": {"open": ["pending"]}}) with pytest.raises(ValueError, match="not declared in fields"): derive_pack_composition(req, "p78ot2o6") def test_workflow_field_wrong_type_is_a_hard_failure(): req = _ticket(workflow={"field": "subject", "transitions": {"open": ["pending"]}}) with pytest.raises(ValueError, match="must be a select field"): derive_pack_composition(req, "p78ot2o6") def test_workflow_transition_referencing_unknown_state_is_a_hard_failure(): req = _ticket(workflow={"field": "status", "transitions": {"open": ["archived"]}}) with pytest.raises(ValueError, match="is not one of"): derive_pack_composition(req, "p78ot2o6") def test_duplicate_transition_targets_are_deduped_deterministically(): """Duplicate targets carry no enforcement meaning (Cloud only checks transitions[from].includes(to)) but must not leak non-deterministically into the schema pushed to Cloud -- see contract.py's validate_workflow_transitions.""" req = _ticket(workflow={ "field": "status", "transitions": {"open": ["pending", "pending", "resolved"], "pending": ["open"], "resolved": []}, }) spec = derive_pack_composition(req, "p78ot2o6") assert not isinstance(spec, PrimitiveGap) assert spec.entities[0].workflow.transitions["open"] == ["pending", "resolved"] def test_two_tickets_with_different_transition_graphs_get_different_fingerprints(): """The fingerprint must include the workflow graph -- otherwise a stricter lifecycle for the same actors/fields would collide onto the same synthesized pack as a looser one.""" strict = derive_pack_composition(_ticket(), "p78ot2o6") looser = derive_pack_composition( _ticket(workflow={ "field": "status", "transitions": {"open": ["pending", "resolved"], "pending": ["open", "resolved"], "resolved": []}, }), "p78ot2o6", ) assert strict.requirement_fingerprint != looser.requirement_fingerprint def test_unsupported_requires_entry_returns_primitive_gap_not_a_spec(): req = _quote_request(requires=["approval.quorum"]) result = derive_pack_composition(req, "p78ot2o6") assert isinstance(result, PrimitiveGap) assert "approval.quorum" in result.missing assert result.criticality == "required" # default def test_primitive_gap_carries_an_explicit_optional_criticality_through(): req = _quote_request(requires=["approval.quorum"], criticality="optional") result = derive_pack_composition(req, "p78ot2o6") assert isinstance(result, PrimitiveGap) assert result.criticality == "optional" def test_cloud_entity_dict_matches_the_real_cloud_create_api_shape(): """Regression test for the Step 0 bug: ComposedEntity.model_dump() alone ({key, label, fields, rls_enabled}) does not match what Cloud's real POST /v1/projects/:id/entities expects, so install_pack's entity loop (entity_def.get("entity_key")) silently skipped every composed entity. Confirmed against alembic/versions/099_feature_packs_seed.py's human-authored shape.""" entity = ComposedEntity( key="quote_request", label="Quote Request", fields=[ {"key": "name", "label": "Name", "type": "text", "required": True}, {"key": "priority", "label": "Priority", "type": "select", "options": ["low", "high"]}, ], rls_enabled=True, ) result = _cloud_entity_dict(entity) assert result["entity_key"] == "quote_request" assert result["name"] == "Quote Request" assert result["plural_name"] == "Quote Request" assert result["rls_enabled"] is True assert result["schema"]["fields"] == [ {"key": "name", "name": "Name", "type": "text", "required": True}, {"key": "priority", "name": "Priority", "type": "select", "required": False, "options": ["low", "high"]}, ] # The old, broken shape must not leak through. assert "key" not in result assert "label" not in result assert "fields" not in result def test_cloud_entity_dict_embeds_access_policy_in_schema_when_present(): """access_policy travels the same channel as workflow -- embedded in schema_json, atomic with the entity itself, not a separate synced document (see the plan's transport decision: no "policy declared but not yet synced" state to reason about).""" entity = ComposedEntity( key="quote_request", label="Quote Request", fields=[{"key": "name", "label": "Name", "type": "text", "required": True}], access_policy={ "version": 1, "roles": {"admin": {"read": {"effect": "allow", "scope": "all"}}}, }, ) result = _cloud_entity_dict(entity) assert result["schema"]["access_policy"] == { "version": 1, "roles": {"admin": {"read": {"effect": "allow", "scope": "all"}}}, } def test_cloud_entity_dict_has_no_access_policy_key_when_absent(): entity = ComposedEntity( key="quote_request", label="Quote Request", fields=[{"key": "name", "label": "Name", "type": "text", "required": True}], ) result = _cloud_entity_dict(entity) assert "access_policy" not in result["schema"] def _fake_db(project_feature_pack_lookup=None): execute_result = SimpleNamespace(scalar_one_or_none=lambda: project_feature_pack_lookup) added: list = [] return SimpleNamespace( execute=AsyncMock(return_value=execute_result), add=lambda obj: added.append(obj), added=added, flush=AsyncMock(), ) @pytest.mark.asyncio async def test_unpublished_function_slug_rejects_the_whole_entry(): req = _quote_request(function_slugs=["some-unpublished-fn"]) db = _fake_db() with patch( "src.domain.edge_functions.repository.FunctionLibraryRepository.get_by_slug", AsyncMock(return_value=SimpleNamespace(status="draft")), ): outcome = await compose_and_register_pack(db, object(), uuid4(), "p78ot2o6", req) assert outcome["ok"] is False assert "unknown_or_unpublished_function_slug:some-unpublished-fn" in outcome["errors"] @pytest.mark.asyncio async def test_primitive_gap_is_reported_not_silently_swallowed(): req = _quote_request(requires=["approval.quorum"]) db = _fake_db() outcome = await compose_and_register_pack(db, object(), uuid4(), "p78ot2o6", req) assert outcome["ok"] is False assert outcome["status"] == "primitive_gap" assert "approval.quorum" in outcome["missing"] assert outcome["criticality"] == "required" @pytest.mark.asyncio async def test_valid_spec_delegates_to_the_existing_unmodified_install_pack(): req = _quote_request() db = _fake_db(project_feature_pack_lookup=None) # no existing row -> create project_id = uuid4() install_mock = AsyncMock(return_value={"ok": True, "status": "installed", "errors": []}) with patch( "src.domain.feature_packs.service.PackInstallService.install_pack", install_mock ): outcome = await compose_and_register_pack(db, object(), project_id, "p78ot2o6", req, "build a quote form") install_mock.assert_awaited_once() _, kwargs = install_mock.call_args assert kwargs["project_id"] == project_id assert kwargs["pack_slug"] == "local-p78ot2o6-quote_request" assert kwargs["user_prompt"] == "build a quote form" assert outcome["ok"] is True assert outcome["pack_slug"] == "local-p78ot2o6-quote_request" @pytest.mark.asyncio async def test_existing_pack_lookup_is_scoped_by_project_id(): """Regression test: the fingerprint dedup lookup used to match purely on (requirement_fingerprint, scope='project_local') with no project_id at all, so two different projects composing the same shape would collide on the same row -- the second project would reuse the first project's slug and overwrite its composition/access_policy.""" req = _quote_request() db = _fake_db() install_mock = AsyncMock(return_value={"ok": True, "status": "installed", "errors": []}) with patch("src.domain.feature_packs.service.PackInstallService.install_pack", install_mock): await compose_and_register_pack(db, object(), uuid4(), "p78ot2o6", req) stmt = db.execute.call_args_list[0][0][0] compiled_sql = str(stmt.compile(compile_kwargs={"literal_binds": True})) assert "feature_packs.project_id" in compiled_sql @pytest.mark.asyncio async def test_newly_created_pack_row_is_stamped_with_its_project_id(): req = _quote_request() db = _fake_db(project_feature_pack_lookup=None) # no existing row -> create project_id = uuid4() install_mock = AsyncMock(return_value={"ok": True, "status": "installed", "errors": []}) with patch("src.domain.feature_packs.service.PackInstallService.install_pack", install_mock): await compose_and_register_pack(db, object(), project_id, "p78ot2o6", req) assert len(db.added) == 1 assert db.added[0].project_id == project_id @pytest.mark.asyncio async def test_repeat_composition_reuses_the_same_row_via_fingerprint(): """Idempotent re-run: a second identical requirement must not create quote-request-02 -- find via requirement_fingerprint and reuse it.""" req = _quote_request() existing_pack = SimpleNamespace( slug="local-p78ot2o6-quote_request", composition={}, access_policy={} ) db = _fake_db(project_feature_pack_lookup=existing_pack) install_mock = AsyncMock(return_value={"ok": True, "status": "installed", "errors": []}) with patch( "src.domain.feature_packs.service.PackInstallService.install_pack", install_mock ): await compose_and_register_pack(db, object(), uuid4(), "p78ot2o6", req) # No new row was constructed -- the existing one was mutated and reused. install_mock.assert_awaited_once() _, kwargs = install_mock.call_args assert kwargs["pack_slug"] == "local-p78ot2o6-quote_request"