From ff075d4fa13408b1f046eba829e49cba1d93cbf3 Mon Sep 17 00:00:00 2001 From: Gregory Clunies Date: Thu, 16 Jul 2026 23:15:31 -0700 Subject: [PATCH] feat(grants): add SPCS privilege enums and grant support Add ComputePoolPriv, ImageRepositoryPriv, and ServicePriv and wire them into PRIVS_FOR_RESOURCE_TYPE, which previously mapped COMPUTE_POOL, IMAGE_REPOSITORY, and SERVICE to None. That gap forced manual SQL for SPCS RBAC: GrantedPrivilege.from_grant fell back to untyped strings and priv: ALL silently expanded to an empty list, causing permanent drift. The generic on_ grant dispatch already handles these types, so no changes to grant.py, data_provider.py, identifiers.py, or blueprint.py were needed. YAML grants via on_compute_pool/on_image_repository/ on_service now round-trip through fetch/plan without spurious drift. Covered by per-enum unit tests, ALL-expansion regression tests, a completeness guard so future grantable types can't silently map to None, and a single-cycle live integration round-trip test. Docs updated with SPCS grant examples and the grantable privilege lists. --- docs/resources/grant.md | 23 +++ docs/snowflake-permissions.md | 8 ++ snowcap/privs.py | 30 +++- tests/integration/test_grant_patterns.py | 106 ++++++++++++++ tests/test_grant.py | 64 +++++++++ tests/test_privs.py | 170 +++++++++++++++++++++++ 6 files changed, 398 insertions(+), 3 deletions(-) diff --git a/docs/resources/grant.md b/docs/resources/grant.md index 81dad6d..04181b1 100644 --- a/docs/resources/grant.md +++ b/docs/resources/grant.md @@ -59,6 +59,21 @@ grants: - priv: MONITOR on: cortex search service somedb.someschema.someservice to: search_observability_role + + # SPCS: USAGE on a compute pool + - priv: usage + on_compute_pool: ds_gpu_pool + to_role: data_engineer + + # SPCS: READ on an image repository + - priv: read + on_image_repository: sandbox.spcs.vllm_repo + to_role: data_engineer + + # SPCS: MONITOR on a service + - priv: monitor + on_service: sandbox.spcs.lora_service + to_role: data_engineer ``` #### Future Grants @@ -115,6 +130,11 @@ grant = Grant(priv="CREATE TABLE", on_schema="foo", to="somerole") # Table Privileges: grant = Grant(priv=["SELECT", "INSERT", "DELETE"], on_table="sometable", to="somerole") + +# Snowpark Container Services (SPCS) Privileges: +grant = Grant(priv="USAGE", on_compute_pool="ds_gpu_pool", to="data_engineer") +grant = Grant(priv="READ", on_image_repository="sandbox.spcs.vllm_repo", to="data_engineer") +grant = Grant(priv="MONITOR", on_service="sandbox.spcs.lora_service", to="data_engineer") ``` #### Future Grants @@ -185,6 +205,9 @@ grant_on_all = Grant( - `"schema my_db.my_schema"` - for schema privileges - `"warehouse my_wh"` - for warehouse privileges - `"database my_db"` - for database privileges + - `"compute pool my_pool"` - for compute pool privileges + - `"image repository my_db.my_schema.my_repo"` - for image repository privileges + - `"service my_db.my_schema.my_service"` - for service privileges - `"future tables in schema my_schema"` - for future grants - `"all tables in database my_db"` - for grants on all existing objects diff --git a/docs/snowflake-permissions.md b/docs/snowflake-permissions.md index 7d9ec22..d3fe905 100644 --- a/docs/snowflake-permissions.md +++ b/docs/snowflake-permissions.md @@ -60,6 +60,14 @@ GRANT CREATE COMPUTE POOL ON ACCOUNT TO ROLE SNOWCAP_ADMIN; GRANT IMPORTED PRIVILEGES ON DATABASE SNOWFLAKE TO ROLE SNOWCAP_ADMIN; ``` +`CREATE COMPUTE POOL` above only covers creating new compute pools. Snowcap can also manage object-level grants on existing SPCS resources (see [Grant](resources/grant.md)): + +| Resource | Grantable privileges | +|---|---| +| Compute pool | `USAGE`, `MONITOR`, `MODIFY`, `OPERATE`, `OWNERSHIP` | +| Image repository | `READ`, `WRITE`, `OWNERSHIP` | +| Service | `USAGE`, `MONITOR`, `OPERATE`, `OWNERSHIP` | + See [Optimizing Grant Fetching with ACCOUNT_USAGE](getting-started.md#optimizing-grant-fetching-with-account_usage) for when the ACCOUNT_USAGE optimization is worth enabling. ### Step 3: Create the service user diff --git a/snowcap/privs.py b/snowcap/privs.py index 4d6ca8c..05bc53b 100644 --- a/snowcap/privs.py +++ b/snowcap/privs.py @@ -112,6 +112,15 @@ class DatabaseRolePriv(Priv): USAGE = "USAGE" +class ComputePoolPriv(Priv): + ALL = "ALL" + MODIFY = "MODIFY" + MONITOR = "MONITOR" + OPERATE = "OPERATE" + OWNERSHIP = "OWNERSHIP" + USAGE = "USAGE" + + class CortexSearchServicePriv(Priv): ALL = "ALL" MONITOR = "MONITOR" @@ -168,6 +177,13 @@ class IcebergTablePriv(Priv): UPDATE = "UPDATE" +class ImageRepositoryPriv(Priv): + ALL = "ALL" + OWNERSHIP = "OWNERSHIP" + READ = "READ" + WRITE = "WRITE" + + class IntegrationPriv(Priv): ALL = "ALL" USAGE = "USAGE" @@ -298,6 +314,14 @@ class SequencePriv(Priv): USAGE = "USAGE" +class ServicePriv(Priv): + ALL = "ALL" + MONITOR = "MONITOR" + OPERATE = "OPERATE" + OWNERSHIP = "OWNERSHIP" + USAGE = "USAGE" + + class StagePriv(Priv): ALL = "ALL" OWNERSHIP = "OWNERSHIP" @@ -391,7 +415,7 @@ class WarehousePriv(Priv): ResourceType.CATALOG_INTEGRATION: None, ResourceType.CLASS: None, ResourceType.COLUMN: None, - ResourceType.COMPUTE_POOL: None, + ResourceType.COMPUTE_POOL: ComputePoolPriv, ResourceType.CORTEX_SEARCH_SERVICE: CortexSearchServicePriv, ResourceType.DATABASE_ROLE: DatabaseRolePriv, ResourceType.DATABASE: DatabasePriv, @@ -408,7 +432,7 @@ class WarehousePriv(Priv): ResourceType.GRANT: None, ResourceType.HYBRID_TABLE: None, ResourceType.ICEBERG_TABLE: IcebergTablePriv, - ResourceType.IMAGE_REPOSITORY: None, + ResourceType.IMAGE_REPOSITORY: ImageRepositoryPriv, ResourceType.INTEGRATION: IntegrationPriv, ResourceType.MATERIALIZED_VIEW: MaterializedViewPriv, ResourceType.NETWORK_POLICY: NetworkPolicyPriv, @@ -428,7 +452,7 @@ class WarehousePriv(Priv): ResourceType.SECRET: SecretPriv, ResourceType.SECURITY_INTEGRATION: IntegrationPriv, ResourceType.SEQUENCE: SequencePriv, - ResourceType.SERVICE: None, + ResourceType.SERVICE: ServicePriv, ResourceType.SHARE: None, ResourceType.STAGE: StagePriv, ResourceType.STORAGE_INTEGRATION: IntegrationPriv, diff --git a/tests/integration/test_grant_patterns.py b/tests/integration/test_grant_patterns.py index 46ab215..0bf072a 100644 --- a/tests/integration/test_grant_patterns.py +++ b/tests/integration/test_grant_patterns.py @@ -246,6 +246,112 @@ def test_no_drift_after_multi_priv_grant_apply(self, cursor, suffix, test_db, ma assert len(plan2) == 0, f"Expected no drift but got: {plan2}" +class TestSPCSGrantsIntegration: + """Test that grants on compute pools, image repositories, and services work correctly.""" + + def test_grants_on_compute_pool_image_repository_and_service( + self, cursor, suffix, test_db, marked_for_cleanup + ): + """Test: USAGE on compute pool + ALL on image repository + ALL on service in one apply. + + Compute pools are slow to provision, so this test creates one compute + pool, one image repository, and one service, then plans and applies + all three SPCS grants in a single blueprint cycle instead of one per + object type. + """ + session = cursor.connection + + role_name = f"SPCS_GRANT_ROLE_{suffix}" + role = res.Role(name=role_name) + cursor.execute(role.create_sql(if_not_exists=True)) + + schema_name = f"SPCS_SCHEMA_{suffix}" + schema = res.Schema(name=f"{test_db}.{schema_name}") + cursor.execute(schema.create_sql(if_not_exists=True)) + + pool_name = f"SPCS_POOL_{suffix}" + cursor.execute( + f"CREATE COMPUTE POOL IF NOT EXISTS {pool_name} " + "MIN_NODES = 1 MAX_NODES = 1 INSTANCE_FAMILY = CPU_X64_XS" + ) + + repo_name = f"SPCS_REPO_{suffix}" + repo_fqn = f"{test_db}.{schema_name}.{repo_name}" + cursor.execute(f"CREATE IMAGE REPOSITORY IF NOT EXISTS {repo_fqn}") + + svc_name = f"SPCS_SVC_{suffix}" + svc_fqn = f"{test_db}.{schema_name}.{svc_name}" + cursor.execute( + f""" +CREATE SERVICE IF NOT EXISTS {svc_fqn} + IN COMPUTE POOL {pool_name} + FROM SPECIFICATION $$ +spec: + container: + - name: main + image: /snowflake/images/snowflake_images/tutorial-image:latest + endpoint: + - name: apiendpoint + port: 8000 + public: true +$$ + MIN_INSTANCES=1 + MAX_INSTANCES=1 +""" + ) + # Cleanup order: service and compute pool are billable and account-scoped + # (outside the DROP DATABASE safety net), so register them first - the + # cleanup loop drops in this append order, and an aborted loop must + # not leak the pool behind an earlier failure. + marked_for_cleanup.append(res.Service(name=svc_fqn, compute_pool=pool_name)) + marked_for_cleanup.append(res.ComputePool(name=pool_name)) + marked_for_cleanup.append(res.ImageRepository(name=repo_fqn)) + marked_for_cleanup.append(schema) + marked_for_cleanup.append(role) + + yaml_config = { + "grants": [ + {"priv": "USAGE", "on_compute_pool": pool_name, "to_role": role_name}, + {"priv": "ALL", "on_image_repository": repo_fqn, "to_role": role_name}, + {"priv": "ALL", "on_service": svc_fqn, "to_role": role_name}, + ], + } + + bc = collect_blueprint_config(yaml_config) + blueprint = Blueprint.from_config(bc) + + plan = blueprint.plan(session) + assert len(plan) == 3 + assert all(isinstance(p, CreateResource) for p in plan) + + blueprint.apply(session, plan) + + # Verify USAGE grant on the compute pool + pool_grant = res.Grant(priv="USAGE", on_compute_pool=pool_name, to=role) + pool_data = safe_fetch(cursor, pool_grant.urn) + assert pool_data is not None, "USAGE grant on compute pool was not created" + assert pool_data["priv"] == "USAGE" + + # Verify ALL grant on the image repository expands to exactly READ, WRITE + repo_grant = res.Grant(priv="ALL", on_image_repository=repo_fqn, to=role) + repo_data = safe_fetch(cursor, repo_grant.urn) + assert repo_data is not None, "ALL grant on image repository was not created" + assert repo_data["_privs"] == ["READ", "WRITE"] + + # Verify ALL grant on the service expands to exactly MONITOR, OPERATE, USAGE + svc_grant = res.Grant(priv="ALL", on_service=svc_fqn, to=role) + svc_data = safe_fetch(cursor, svc_grant.urn) + assert svc_data is not None, "ALL grant on service was not created" + assert svc_data["_privs"] == ["MONITOR", "OPERATE", "USAGE"] + + # Re-plan should show no changes + reset_cache() + bc2 = collect_blueprint_config(yaml_config) + blueprint2 = Blueprint.from_config(bc2) + plan2 = blueprint2.plan(session) + assert len(plan2) == 0, f"Expected no drift but got: {plan2}" + + class TestRoleGrantsIntegration: """Test that role grants work correctly in Snowflake.""" diff --git a/tests/test_grant.py b/tests/test_grant.py index a32961b..07ff758 100644 --- a/tests/test_grant.py +++ b/tests/test_grant.py @@ -174,6 +174,55 @@ def test_grant_on_cortex_search_service(): assert "MONITOR ON CORTEX SEARCH SERVICE" in monitor_grant.create_sql() +def test_grant_on_compute_pool(): + """USAGE on a COMPUTE POOL parses and renders correctly. + + Compute pools are account-scoped, so grants target a bare name: + GRANT USAGE ON COMPUTE POOL TO ROLE r + """ + grant = res.Grant( + priv="USAGE", + on_compute_pool="MY_POOL", + to="SOME_ROLE", + ) + assert grant.on_type == ResourceType.COMPUTE_POOL + assert grant.on == "MY_POOL" + assert "GRANT USAGE ON COMPUTE POOL" in grant.create_sql() + + +def test_grant_on_image_repository(): + """READ on an IMAGE REPOSITORY parses and renders correctly. + + Image repositories are schema-scoped, so grants target a fully + qualified name: + GRANT READ ON IMAGE REPOSITORY .. TO ROLE r + """ + grant = res.Grant( + priv="READ", + on_image_repository="DB.SPCS.REPO", + to="SOME_ROLE", + ) + assert grant.on_type == ResourceType.IMAGE_REPOSITORY + assert grant.on == "DB.SPCS.REPO" + assert "GRANT READ ON IMAGE REPOSITORY DB.SPCS.REPO" in grant.create_sql() + + +def test_grant_on_service(): + """MONITOR on a SERVICE parses and renders correctly. + + Services are schema-scoped, so grants target a fully qualified name: + GRANT MONITOR ON SERVICE .. TO ROLE r + """ + grant = res.Grant( + priv="MONITOR", + on_service="DB.SPCS.SVC", + to="SOME_ROLE", + ) + assert grant.on_type == ResourceType.SERVICE + assert grant.on == "DB.SPCS.SVC" + assert "GRANT MONITOR ON SERVICE DB.SPCS.SVC" in grant.create_sql() + + def test_grant_database_role_to_database_role(): database = res.Database(name="somedb") parent = res.DatabaseRole(name="parent", database=database) @@ -328,6 +377,21 @@ def test_multi_priv_empty_list_raises_error(self): with pytest.raises(ValueError, match="at least one privilege"): res.Grant(priv=[], on_warehouse="somewh", to="somerole") + def test_all_priv_expands_on_compute_pool(self): + """Test: priv: ALL on a compute pool expands to the full non-ALL/non-OWNERSHIP list.""" + grant = res.Grant(priv="ALL", on_compute_pool="MY_POOL", to="R") + assert grant._data._privs == ["MODIFY", "MONITOR", "OPERATE", "USAGE"] + + def test_all_priv_expands_on_image_repository(self): + """Test: priv: ALL on an image repository expands to the full non-ALL/non-OWNERSHIP list.""" + grant = res.Grant(priv="ALL", on_image_repository="D.S.REPO", to="R") + assert grant._data._privs == ["READ", "WRITE"] + + def test_all_priv_expands_on_service(self): + """Test: priv: ALL on a service expands to the full non-ALL/non-OWNERSHIP list.""" + grant = res.Grant(priv="ALL", on_service="D.S.SVC", to="R") + assert grant._data._privs == ["MONITOR", "OPERATE", "USAGE"] + def test_single_priv_in_list_works(self): """Test: Single privilege in list works correctly.""" grant = res.Grant(priv=["USAGE"], on_warehouse="somewh", to="somerole") diff --git a/tests/test_privs.py b/tests/test_privs.py index 6450721..8ecb84f 100644 --- a/tests/test_privs.py +++ b/tests/test_privs.py @@ -7,6 +7,7 @@ Priv, AccountPriv, AlertPriv, + ComputePoolPriv, CortexSearchServicePriv, DatabasePriv, DatabaseRolePriv, @@ -17,6 +18,7 @@ FileFormatPriv, FunctionPriv, IcebergTablePriv, + ImageRepositoryPriv, IntegrationPriv, MaterializedViewPriv, NetworkPolicyPriv, @@ -31,6 +33,7 @@ SchemaPriv, SecretPriv, SequencePriv, + ServicePriv, StagePriv, StreamPriv, StreamlitPriv, @@ -68,6 +71,38 @@ def test_resource_privs_is_complete(): assert resource_type in PRIVS_FOR_RESOURCE_TYPE, f"{resource_type} is missing from PRIVS_FOR_RESOURCE_TYPE" +# Resource types deliberately mapped to None in PRIVS_FOR_RESOURCE_TYPE: no grantable Priv +# enum exists for them yet (or, for genuinely non-grantable pseudo-types, ever will). Any other +# ResourceType mapped to None would silently break `priv: ALL` expansion (empty _privs), so +# test_resource_privs_maps_to_priv_enum requires new entries to be added here explicitly. +RESOURCE_TYPES_WITHOUT_PRIV_ENUM = { + ResourceType.AGGREGATION_POLICY, + ResourceType.APPLICATION_ROLE, + ResourceType.AUTHENTICATION_POLICY, + ResourceType.CATALOG_INTEGRATION, + ResourceType.CLASS, + ResourceType.COLUMN, + ResourceType.GRANT, + ResourceType.HYBRID_TABLE, + ResourceType.RESOURCE_MONITOR, + ResourceType.ROLE_GRANT, + ResourceType.SHARE, + ResourceType.TAG_MASKING_POLICY_REFERENCE, + ResourceType.TAG_REFERENCE, +} + + +def test_resource_privs_maps_to_priv_enum(): + for resource_type, priv_enum in PRIVS_FOR_RESOURCE_TYPE.items(): + if resource_type in RESOURCE_TYPES_WITHOUT_PRIV_ENUM: + continue + assert priv_enum is not None, ( + f"{resource_type} maps to None in PRIVS_FOR_RESOURCE_TYPE; either add a Priv subclass " + "or add it to RESOURCE_TYPES_WITHOUT_PRIV_ENUM if it's genuinely non-grantable" + ) + assert issubclass(priv_enum, Priv), f"{resource_type} maps to {priv_enum}, which is not a Priv subclass" + + ############################################################################# # Priv Base Class Tests ############################################################################# @@ -87,6 +122,7 @@ def test_priv_subclasses(self): priv_classes = [ AccountPriv, AlertPriv, + ComputePoolPriv, DatabasePriv, DatabaseRolePriv, DirectoryTablePriv, @@ -96,6 +132,7 @@ def test_priv_subclasses(self): FileFormatPriv, FunctionPriv, IcebergTablePriv, + ImageRepositoryPriv, IntegrationPriv, MaterializedViewPriv, NetworkPolicyPriv, @@ -110,6 +147,7 @@ def test_priv_subclasses(self): SchemaPriv, SecretPriv, SequencePriv, + ServicePriv, StagePriv, StreamPriv, StreamlitPriv, @@ -299,6 +337,93 @@ def test_modify_privilege(self): assert WarehousePriv.MODIFY.value == "MODIFY" +############################################################################# +# ComputePoolPriv Tests +############################################################################# + + +class TestComputePoolPriv: + """Tests for ComputePoolPriv enum values.""" + + def test_all_privilege(self): + """ComputePoolPriv has ALL privilege.""" + assert ComputePoolPriv.ALL.value == "ALL" + + def test_modify_privilege(self): + """ComputePoolPriv has MODIFY privilege.""" + assert ComputePoolPriv.MODIFY.value == "MODIFY" + + def test_monitor_privilege(self): + """ComputePoolPriv has MONITOR privilege.""" + assert ComputePoolPriv.MONITOR.value == "MONITOR" + + def test_operate_privilege(self): + """ComputePoolPriv has OPERATE privilege.""" + assert ComputePoolPriv.OPERATE.value == "OPERATE" + + def test_ownership_privilege(self): + """ComputePoolPriv has OWNERSHIP privilege.""" + assert ComputePoolPriv.OWNERSHIP.value == "OWNERSHIP" + + def test_usage_privilege(self): + """ComputePoolPriv has USAGE privilege.""" + assert ComputePoolPriv.USAGE.value == "USAGE" + + +############################################################################# +# ImageRepositoryPriv Tests +############################################################################# + + +class TestImageRepositoryPriv: + """Tests for ImageRepositoryPriv enum values.""" + + def test_all_privilege(self): + """ImageRepositoryPriv has ALL privilege.""" + assert ImageRepositoryPriv.ALL.value == "ALL" + + def test_ownership_privilege(self): + """ImageRepositoryPriv has OWNERSHIP privilege.""" + assert ImageRepositoryPriv.OWNERSHIP.value == "OWNERSHIP" + + def test_read_privilege(self): + """ImageRepositoryPriv has READ privilege.""" + assert ImageRepositoryPriv.READ.value == "READ" + + def test_write_privilege(self): + """ImageRepositoryPriv has WRITE privilege.""" + assert ImageRepositoryPriv.WRITE.value == "WRITE" + + +############################################################################# +# ServicePriv Tests +############################################################################# + + +class TestServicePriv: + """Tests for ServicePriv enum values.""" + + def test_all_privilege(self): + """ServicePriv has ALL privilege.""" + assert ServicePriv.ALL.value == "ALL" + + def test_monitor_privilege(self): + """ServicePriv has MONITOR privilege.""" + assert ServicePriv.MONITOR.value == "MONITOR" + + def test_operate_privilege(self): + """ServicePriv has OPERATE privilege.""" + assert ServicePriv.OPERATE.value == "OPERATE" + + def test_ownership_privilege(self): + """ServicePriv has OWNERSHIP privilege.""" + assert ServicePriv.OWNERSHIP.value == "OWNERSHIP" + + def test_usage_privilege(self): + """ServicePriv has USAGE privilege.""" + assert ServicePriv.USAGE.value == "USAGE" + + ############################################################################# # StagePriv Tests ############################################################################# @@ -430,6 +555,21 @@ def test_warehouse_privs(self): assert "ALL" not in privs assert "OWNERSHIP" not in privs + def test_compute_pool_privs(self): + """all_privs_for_resource_type returns non-ALL/OWNERSHIP privileges for compute pool.""" + privs = all_privs_for_resource_type(ResourceType.COMPUTE_POOL) + assert privs == ["MODIFY", "MONITOR", "OPERATE", "USAGE"] + + def test_image_repository_privs(self): + """all_privs_for_resource_type returns non-ALL/OWNERSHIP privileges for image repository.""" + privs = all_privs_for_resource_type(ResourceType.IMAGE_REPOSITORY) + assert privs == ["READ", "WRITE"] + + def test_service_privs(self): + """all_privs_for_resource_type returns non-ALL/OWNERSHIP privileges for service.""" + privs = all_privs_for_resource_type(ResourceType.SERVICE) + assert privs == ["MONITOR", "OPERATE", "USAGE"] + ############################################################################# # system_role_for_priv() Tests @@ -511,6 +651,24 @@ def test_from_grant_with_schema(self): assert gp.privilege == SchemaPriv.USAGE assert gp.on == "MY_SCHEMA" + def test_from_grant_with_compute_pool(self): + """GrantedPrivilege.from_grant works with compute pool resource type.""" + gp = GrantedPrivilege.from_grant(privilege="USAGE", granted_on="COMPUTE POOL", name="MY_POOL") + assert gp.privilege == ComputePoolPriv.USAGE + assert gp.on == "MY_POOL" + + def test_from_grant_with_image_repository(self): + """GrantedPrivilege.from_grant works with image repository resource type.""" + gp = GrantedPrivilege.from_grant(privilege="READ", granted_on="IMAGE REPOSITORY", name="MY_REPO") + assert gp.privilege == ImageRepositoryPriv.READ + assert gp.on == "MY_REPO" + + def test_from_grant_with_service(self): + """GrantedPrivilege.from_grant works with service resource type.""" + gp = GrantedPrivilege.from_grant(privilege="MONITOR", granted_on="SERVICE", name="MY_SERVICE") + assert gp.privilege == ServicePriv.MONITOR + assert gp.on == "MY_SERVICE" + def test_from_grant_with_no_priv_type(self): """GrantedPrivilege.from_grant returns string privilege for resource types without privilege enums.""" # GRANT resource type has no associated privilege enum @@ -570,6 +728,18 @@ def test_grant_maps_to_none(self): """GRANT resource type maps to None (pseudo-resource).""" assert PRIVS_FOR_RESOURCE_TYPE[ResourceType.GRANT] is None + def test_compute_pool_maps_to_compute_pool_priv(self): + """COMPUTE_POOL resource type maps to ComputePoolPriv.""" + assert PRIVS_FOR_RESOURCE_TYPE[ResourceType.COMPUTE_POOL] is ComputePoolPriv + + def test_image_repository_maps_to_image_repository_priv(self): + """IMAGE_REPOSITORY resource type maps to ImageRepositoryPriv.""" + assert PRIVS_FOR_RESOURCE_TYPE[ResourceType.IMAGE_REPOSITORY] is ImageRepositoryPriv + + def test_service_maps_to_service_priv(self): + """SERVICE resource type maps to ServicePriv.""" + assert PRIVS_FOR_RESOURCE_TYPE[ResourceType.SERVICE] is ServicePriv + def test_integration_types_map_to_integration_priv(self): """Integration resource types map to IntegrationPriv.""" assert PRIVS_FOR_RESOURCE_TYPE[ResourceType.API_INTEGRATION] == IntegrationPriv