fix(lifecycle): transfer ownership with the object type GRANT OWNERSHIP accepts

snowcap/enums.py

@@ -397,6 +397,11 @@ def resource_type_is_grant(resource_type: ResourceType) -> bool:
     )
 
 
+def resource_type_is_integration(resource_type: str) -> bool:
+    """Return True for INTEGRATION and every integration subtype, such as SECURITY INTEGRATION."""
+    return "INTEGRATION" in str(resource_type)
+
+
 # Object types whose body or schedule runs with the privileges of their owner rather than
 # the caller. Snowflake requires stricter authorization to transfer ownership of these:
 # the receiving role must be in the caller's active role hierarchy, or the caller must hold

snowcap/lifecycle.py

@@ -4,7 +4,7 @@
 from inflection import pluralize
 
 from .builder import tidy_sql
-from .enums import GrantType, ResourceType
+from .enums import GrantType, ResourceType, resource_type_is_integration
 from .identifiers import FQN, URN
 from .props import BoolProp, IntProp, Props, StringProp
 from .resource_name import ResourceName
@@ -174,7 +174,7 @@ def _grant_container_sql(data: dict) -> str:
 
 def create_grant(urn: URN, data: dict, props: Props, if_not_exists: bool):
     on_type = data["on_type"]
-    if "INTEGRATION" in str(on_type):
+    if resource_type_is_integration(on_type):
         on_type = "INTEGRATION"
     elif on_type == "ACCOUNT":
         on_type = ""
@@ -193,7 +193,7 @@ def create_grant(urn: URN, data: dict, props: Props, if_not_exists: bool):
         )
     if data["grant_type"] == GrantType.FUTURE:
         items_type = data["items_type"]
-        if "INTEGRATION" in items_type:
+        if resource_type_is_integration(items_type):
             items_type = "INTEGRATION"
         return tidy_sql(
             "GRANT",
@@ -210,7 +210,7 @@ def create_grant(urn: URN, data: dict, props: Props, if_not_exists: bool):
         )
     elif data["grant_type"] == GrantType.ALL:
         items_type = data["items_type"]
-        if "INTEGRATION" in items_type:
+        if resource_type_is_integration(items_type):
             items_type = "INTEGRATION"
         return tidy_sql(
             "GRANT",
@@ -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 resource_type_is_integration(resource_type):
+        return "INTEGRATION"
+    return _OWNERSHIP_OBJECT_TYPES.get(resource_type, str(resource_type))
+
+
 def transfer__default(
     urn: URN,
     owner: str,
@@ -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,

snowcap/resources/grant.py

@@ -10,6 +10,7 @@
     GrantType,
     ParseableEnum,
     ResourceType,
+    resource_type_is_integration,
 )
 from ..identifiers import (
     FQN,
@@ -480,7 +481,7 @@ def __init__(
             # Hacky fix
             if on_type == ResourceType.SCHEMA and on.upper().startswith("SNOWFLAKE"):
                 owner = "ACCOUNTADMIN"
-            elif "INTEGRATION" in str(on_type):
+            elif resource_type_is_integration(on_type):
                 owner = "ACCOUNTADMIN"
             else:
                 owner = "SYSADMIN"

tests/test_lifecycle.py

@@ -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