Skip to content
Open
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
19 changes: 18 additions & 1 deletion snowcap/lifecycle.py
Original file line number Diff line number Diff line change
Expand Up @@ -868,6 +868,23 @@ def transfer_resource(
)


# GRANT OWNERSHIP names these resource types by a broader object type. Snowflake
# rejects EXTERNAL FUNCTION and the integration subtype names, and its usage notes
# say to use VIEW for materialized views and TABLE for hybrid tables.
# https://docs.snowflake.com/en/sql-reference/sql/grant-ownership
_OWNERSHIP_OBJECT_TYPES = {
ResourceType.EXTERNAL_FUNCTION: "FUNCTION",
ResourceType.HYBRID_TABLE: "TABLE",
ResourceType.MATERIALIZED_VIEW: "VIEW",
}


def _ownership_object_type(resource_type: ResourceType) -> str:
if "INTEGRATION" in str(resource_type):
return "INTEGRATION"
return _OWNERSHIP_OBJECT_TYPES.get(resource_type, str(resource_type))


def transfer__default(
urn: URN,
owner: str,
Expand All @@ -877,7 +894,7 @@ def transfer__default(
) -> str:
return tidy_sql(
"GRANT OWNERSHIP ON",
urn.resource_type,
_ownership_object_type(urn.resource_type),
urn.fqn,
"TO",
owner_resource_type,
Expand Down
29 changes: 29 additions & 0 deletions tests/test_lifecycle.py
Original file line number Diff line number Diff line change
Expand Up @@ -1316,6 +1316,35 @@ def test_transfer_default(self):
# ResourceType.DATABASE_ROLE renders as "DATABASE ROLE"
assert "TO DATABASE ROLE NEW_OWNER" in result

@pytest.mark.parametrize(
"resource_type,schema,object_type",
[
(ResourceType.API_INTEGRATION, None, "INTEGRATION"),
(ResourceType.CATALOG_INTEGRATION, None, "INTEGRATION"),
(ResourceType.EXTERNAL_ACCESS_INTEGRATION, None, "INTEGRATION"),
(ResourceType.NOTIFICATION_INTEGRATION, None, "INTEGRATION"),
(ResourceType.SECURITY_INTEGRATION, None, "INTEGRATION"),
(ResourceType.STORAGE_INTEGRATION, None, "INTEGRATION"),
(ResourceType.EXTERNAL_FUNCTION, "MY_SCHEMA", "FUNCTION"),
(ResourceType.HYBRID_TABLE, "MY_SCHEMA", "TABLE"),
(ResourceType.MATERIALIZED_VIEW, "MY_SCHEMA", "VIEW"),
# Types close to the mapped ones keep their own names.
(ResourceType.DYNAMIC_TABLE, "MY_SCHEMA", "DYNAMIC TABLE"),
(ResourceType.ICEBERG_TABLE, "MY_SCHEMA", "ICEBERG TABLE"),
(ResourceType.VIEW, "MY_SCHEMA", "VIEW"),
],
)
def test_transfer_uses_snowflake_ownership_object_type(self, resource_type, schema, object_type):
"""GRANT OWNERSHIP names some resource types by a broader object type.

https://docs.snowflake.com/en/sql-reference/sql/grant-ownership
"""
database = "MY_DB" if schema else None
urn = make_urn(resource_type, "MY_OBJECT", database=database, schema=schema)
result = transfer_resource(urn, "NEW_OWNER", ResourceType.ROLE, copy_current_grants=True)
assert result.startswith(f"GRANT OWNERSHIP ON {object_type} ")
assert "MY_OBJECT TO ROLE NEW_OWNER COPY CURRENT GRANTS" in result


# ============================================================================
# Test tag masking policy reference functions
Expand Down
Loading