fix(grant): stop phantom CREATE for database-role-to-database-role grants

snowcap/data_provider.py

@@ -1524,7 +1524,10 @@ def _fetch_grants_from_account_usage(session: SnowflakeConnection) -> list[dict[
 
         # Construct fully qualified name to match SHOW GRANTS output
         # ACCOUNT_USAGE NAME column only has object name, not full path
+        # Same 'DATABASE_ROLE' -> 'DATABASE ROLE' normalization as granted_to above
         granted_on = row["GRANTED_ON"]
+        if granted_on == "DATABASE_ROLE":
+            granted_on = "DATABASE ROLE"
         name = row["NAME"]
         table_catalog = row.get("TABLE_CATALOG")
         table_schema = row.get("TABLE_SCHEMA")
@@ -1539,7 +1542,7 @@ def _fetch_grants_from_account_usage(session: SnowflakeConnection) -> list[dict[
             # Schema grants: need DATABASE.SCHEMA
             if table_catalog:
                 name = f"{table_catalog}.{name}"
-        elif granted_on in ("DATABASE ROLE",):
+        elif granted_on == "DATABASE ROLE":
             # Database role grants: need DATABASE.ROLE
             if table_catalog:
                 name = f"{table_catalog}.{name}"
@@ -2106,12 +2109,37 @@ def fetch_database_role(session: SnowflakeConnection, fqn: FQN):
     }
 
 
+def _database_role_grantee_fqn(grantee_name: str, default_database: Optional[ResourceName]) -> FQN:
+    """
+    Build an FQN for a DATABASE_ROLE grant's grantee.
+
+    Snowflake reports a database-role grantee unqualified when it lives in the same
+    database as the role granting it, but fully qualified when it's in another database.
+    Parsing it as an FQN and filling in default_database only when it's missing makes both
+    cases comparable to DatabaseRoleGrant.to_database_role's FQN, which is always fully
+    qualified.
+    """
+    grantee_fqn = parse_FQN(grantee_name, is_db_scoped=True)
+    if grantee_fqn.database is None:
+        grantee_fqn.database = default_database
+    return grantee_fqn
+
+
 def fetch_database_role_grant(session: SnowflakeConnection, fqn: FQN):
     show_result = execute(session, f"SHOW GRANTS OF DATABASE ROLE {fqn.database}.{fqn.name}", cacheable=True)
 
     subject, subject_name = next(iter(fqn.params.items()))
 
-    role_grants = _filter_result(show_result, granted_to=subject.upper(), grantee_name=subject_name)
+    if subject == "database_role":
+        target = parse_FQN(subject_name, is_db_scoped=True)
+        role_grants = [
+            row
+            for row in _filter_result(show_result, granted_to=subject.upper())
+            if _database_role_grantee_fqn(row["grantee_name"], fqn.database) == target
+        ]
+    else:
+        role_grants = _filter_result(show_result, granted_to=subject.upper(), grantee_name=subject_name)
+
     if len(role_grants) == 0:
         return None
     if len(role_grants) > 1:
@@ -2124,7 +2152,7 @@ def fetch_database_role_grant(session: SnowflakeConnection, fqn: FQN):
     if data["granted_to"] == "ROLE":
         to_role = _quote_snowflake_identifier(data["grantee_name"])
     elif data["granted_to"] == "DATABASE_ROLE":
-        to_database_role = data["grantee_name"]
+        to_database_role = str(_database_role_grantee_fqn(data["grantee_name"], fqn.database))
 
     return {
         "database_role": data["role"],
@@ -4279,12 +4307,17 @@ def list_database_role_grants(
 
                 # Determine subject based on grantee type
                 subject = "role" if grant["granted_to"] == "ROLE" else "database_role"
+                grantee_name = grant["grantee_name"]
+                if subject == "database_role":
+                    grantee_name = str(
+                        _database_role_grantee_fqn(grantee_name, resource_name_from_snowflake_metadata(db_name))
+                    )
 
                 role_grants.append(
                     FQN(
                         name=resource_name_from_snowflake_metadata(role_name),
                         database=resource_name_from_snowflake_metadata(db_name),
-                        params={subject: grant["grantee_name"]},
+                        params={subject: grantee_name},
                     )
                 )
             # If we got results or no specific database was filtered, return
@@ -4318,11 +4351,16 @@ def list_database_role_grants(
             for data in show_result:
                 subject = "role" if data["granted_to"] == "ROLE" else "database_role"
                 db, name = data["role"].split(".")
+                grantee_name = data["grantee_name"]
+                if subject == "database_role":
+                    grantee_name = str(
+                        _database_role_grantee_fqn(grantee_name, resource_name_from_snowflake_metadata(db))
+                    )
                 role_grants.append(
                     FQN(
                         name=resource_name_from_snowflake_metadata(name),
                         database=resource_name_from_snowflake_metadata(db),
-                        params={subject: data["grantee_name"]},
+                        params={subject: grantee_name},
                     )
                 )
     return role_grants

tests/test_data_provider.py

@@ -2490,6 +2490,251 @@ def execute_side_effect(session, query, **kwargs):
         assert not [on for on in on_values if "database_role" in on]
 
 
+class TestDatabaseRoleGrantToDatabaseRole:
+    """Snowflake reports a DATABASE_ROLE grantee unqualified when it's in the same database
+    as the role granting it, but the manifest side always builds a fully qualified
+    DB.ROLE string. Comparing the two directly never matches, so plan re-issues the grant
+    as a no-op CREATE on every run."""
+
+    @patch("snowcap.data_provider.execute")
+    def test_fetch_matches_unqualified_grantee_in_same_database(self, mock_execute):
+        from snowcap.data_provider import fetch_database_role_grant
+        from snowcap.identifiers import FQN
+        from snowcap.resource_name import ResourceName
+
+        mock_execute.return_value = [
+            {
+                "role": "GREAT_BAY_DEV.DR_READER_ROLE",
+                "granted_to": "DATABASE_ROLE",
+                "grantee_name": "DR_WRITER_ROLE",
+                "granted_by": "SECURITYADMIN",
+            }
+        ]
+        fqn = FQN(
+            name=ResourceName("DR_READER_ROLE"),
+            database=ResourceName("GREAT_BAY_DEV"),
+            params={"database_role": "GREAT_BAY_DEV.DR_WRITER_ROLE"},
+        )
+
+        result = fetch_database_role_grant(MagicMock(), fqn)
+
+        assert result is not None
+        assert result["to_database_role"] == "GREAT_BAY_DEV.DR_WRITER_ROLE"
+
+    @patch("snowcap.data_provider.execute")
+    def test_fetch_matches_qualified_grantee_in_another_database(self, mock_execute):
+        from snowcap.data_provider import fetch_database_role_grant
+        from snowcap.identifiers import FQN
+        from snowcap.resource_name import ResourceName
+
+        mock_execute.return_value = [
+            {
+                "role": "GREAT_BAY_DEV.DR_READER_ROLE",
+                "granted_to": "DATABASE_ROLE",
+                "grantee_name": "OTHER_DB.DR_WRITER_ROLE",
+                "granted_by": "SECURITYADMIN",
+            }
+        ]
+        fqn = FQN(
+            name=ResourceName("DR_READER_ROLE"),
+            database=ResourceName("GREAT_BAY_DEV"),
+            params={"database_role": "OTHER_DB.DR_WRITER_ROLE"},
+        )
+
+        result = fetch_database_role_grant(MagicMock(), fqn)
+
+        assert result is not None
+        assert result["to_database_role"] == "OTHER_DB.DR_WRITER_ROLE"
+
+    @patch("snowcap.data_provider.execute")
+    def test_fetch_does_not_false_match_grantee_in_a_different_database(self, mock_execute):
+        """OTHERDB.DR_WRITER_ROLE must not match a target of DB.DR_WRITER_ROLE just because
+        the bare role name is the same."""
+        from snowcap.data_provider import fetch_database_role_grant
+        from snowcap.identifiers import FQN
+        from snowcap.resource_name import ResourceName
+
+        mock_execute.return_value = [
+            {
+                "role": "GREAT_BAY_DEV.DR_READER_ROLE",
+                "granted_to": "DATABASE_ROLE",
+                "grantee_name": "OTHER_DB.DR_WRITER_ROLE",
+                "granted_by": "SECURITYADMIN",
+            }
+        ]
+        fqn = FQN(
+            name=ResourceName("DR_READER_ROLE"),
+            database=ResourceName("GREAT_BAY_DEV"),
+            params={"database_role": "GREAT_BAY_DEV.DR_WRITER_ROLE"},
+        )
+
+        assert fetch_database_role_grant(MagicMock(), fqn) is None
+
+    @patch("snowcap.data_provider.execute")
+    def test_fetch_regression_grant_to_account_role(self, mock_execute):
+        from snowcap.data_provider import fetch_database_role_grant
+        from snowcap.identifiers import FQN
+        from snowcap.resource_name import ResourceName
+
+        mock_execute.return_value = [
+            {
+                "role": "GREAT_BAY_DEV.DR_READER_ROLE",
+                "granted_to": "ROLE",
+                "grantee_name": "GREAT_BAY_DEV__READER",
+                "granted_by": "SECURITYADMIN",
+            }
+        ]
+        fqn = FQN(
+            name=ResourceName("DR_READER_ROLE"),
+            database=ResourceName("GREAT_BAY_DEV"),
+            params={"role": "GREAT_BAY_DEV__READER"},
+        )
+
+        result = fetch_database_role_grant(MagicMock(), fqn)
+
+        assert result is not None
+        assert result["to_role"] == "GREAT_BAY_DEV__READER"
+
+    @patch("snowcap.data_provider._should_use_account_usage")
+    @patch("snowcap.data_provider.execute")
+    def test_list_show_path_qualifies_same_database_grantee(self, mock_execute, mock_should_use):
+        from snowcap.data_provider import list_database_role_grants
+
+        mock_should_use.return_value = False
+
+        def execute_side_effect(session, query, **kwargs):
+            if "SHOW DATABASE ROLES IN DATABASE" in query:
+                return [{"name": "DR_READER_ROLE"}]
+            if "SHOW GRANTS OF DATABASE ROLE" in query:
+                return [
+                    {
+                        "role": "GREAT_BAY_DEV.DR_READER_ROLE",
+                        "granted_to": "DATABASE_ROLE",
+                        "grantee_name": "DR_WRITER_ROLE",
+                        "granted_by": "SECURITYADMIN",
+                    }
+                ]
+            raise AssertionError(f"unexpected query: {query}")
+
+        mock_execute.side_effect = execute_side_effect
+
+        grants = list_database_role_grants(MagicMock(), database="GREAT_BAY_DEV")
+
+        assert len(grants) == 1
+        assert grants[0].params == {"database_role": "GREAT_BAY_DEV.DR_WRITER_ROLE"}
+
+    @patch("snowcap.data_provider._should_use_account_usage")
+    @patch("snowcap.data_provider.execute")
+    def test_list_account_usage_path_recognizes_underscore_granted_on(self, mock_execute, mock_should_use):
+        """SNOWFLAKE.ACCOUNT_USAGE.GRANTS_TO_ROLES spells the object type GRANTED_ON =
+        'DATABASE_ROLE' (underscore). _fetch_grants_from_account_usage must normalize that
+        to 'DATABASE ROLE' (matching SHOW GRANTS) or list_database_role_grants's filter on
+        granted_on never matches and this path always returns zero results."""
+        from datetime import datetime
+
+        from snowcap.data_provider import list_database_role_grants
+
+        mock_should_use.return_value = True
+        mock_execute.return_value = [
+            {
+                "CREATED_ON": datetime(2024, 1, 1),
+                "PRIVILEGE": "USAGE",
+                "GRANTED_ON": "DATABASE_ROLE",
+                "NAME": "DR_READER_ROLE",
+                "TABLE_CATALOG": "GREAT_BAY_DEV",
+                "GRANTED_TO": "DATABASE_ROLE",
+                "GRANTEE_NAME": "DR_WRITER_ROLE",
+                "GRANT_OPTION": False,
+                "GRANTED_BY": "SECURITYADMIN",
+            }
+        ]
+
+        grants = list_database_role_grants(MagicMock(), database="GREAT_BAY_DEV", use_account_usage=True)
+
+        assert len(grants) == 1
+        assert grants[0].params == {"database_role": "GREAT_BAY_DEV.DR_WRITER_ROLE"}
+
+    def test_account_usage_and_show_paths_return_identical_fqns(self):
+        """The two list paths are interchangeable sources for the same diff, so they must
+        agree on the FQN they produce for the same underlying grant."""
+        from snowcap.data_provider import list_database_role_grants
+
+        au_row = {
+            "privilege": "USAGE",
+            "granted_on": "DATABASE ROLE",  # as normalized by _fetch_grants_from_account_usage
+            "name": "GREAT_BAY_DEV.DR_READER_ROLE",
+            "granted_to": "DATABASE_ROLE",
+            "grantee_name": "DR_WRITER_ROLE",
+        }
+        show_row = {
+            "role": "GREAT_BAY_DEV.DR_READER_ROLE",
+            "granted_to": "DATABASE_ROLE",
+            "grantee_name": "DR_WRITER_ROLE",
+            "granted_by": "SECURITYADMIN",
+        }
+
+        with (
+            patch("snowcap.data_provider._should_use_account_usage", return_value=True),
+            patch("snowcap.data_provider._fetch_grants_from_account_usage", return_value=[au_row]),
+        ):
+            au_grants = list_database_role_grants(MagicMock(), database="GREAT_BAY_DEV", use_account_usage=True)
+
+        def show_side_effect(session, query, **kwargs):
+            if "SHOW DATABASE ROLES IN DATABASE" in query:
+                return [{"name": "DR_READER_ROLE"}]
+            if "SHOW GRANTS OF DATABASE ROLE" in query:
+                return [show_row]
+            raise AssertionError(f"unexpected query: {query}")
+
+        with (
+            patch("snowcap.data_provider._should_use_account_usage", return_value=False),
+            patch("snowcap.data_provider.execute", side_effect=show_side_effect),
+        ):
+            show_grants = list_database_role_grants(MagicMock(), database="GREAT_BAY_DEV")
+
+        assert len(au_grants) == 1
+        assert len(show_grants) == 1
+        assert au_grants[0] == show_grants[0]
+        assert au_grants[0].params == {"database_role": "GREAT_BAY_DEV.DR_WRITER_ROLE"}
+
+    def test_fetched_grant_matches_declared_manifest_fqn(self):
+        """The actual reported symptom: plan compares the fetched FQN against the FQN the
+        manifest builds for the same declared grant. They must be equal, or plan proposes a
+        CREATE for a grant that already exists."""
+        from snowcap import resources as res
+        from snowcap.resources.grant import database_role_grant_fqn
+        from snowcap.data_provider import list_database_role_grants
+
+        grant = res.DatabaseRoleGrant(
+            database_role="great_bay_dev.dr_reader_role",
+            to_database_role="great_bay_dev.dr_writer_role",
+        )
+        declared_fqn = database_role_grant_fqn(grant._data)
+
+        def show_side_effect(session, query, **kwargs):
+            if "SHOW DATABASE ROLES IN DATABASE" in query:
+                return [{"name": "DR_READER_ROLE"}]
+            if "SHOW GRANTS OF DATABASE ROLE" in query:
+                return [
+                    {
+                        "role": "GREAT_BAY_DEV.DR_READER_ROLE",
+                        "granted_to": "DATABASE_ROLE",
+                        "grantee_name": "DR_WRITER_ROLE",
+                        "granted_by": "SECURITYADMIN",
+                    }
+                ]
+            raise AssertionError(f"unexpected query: {query}")
+
+        with (
+            patch("snowcap.data_provider._should_use_account_usage", return_value=False),
+            patch("snowcap.data_provider.execute", side_effect=show_side_effect),
+        ):
+            fetched_grants = list_database_role_grants(MagicMock(), database="GREAT_BAY_DEV")
+
+        assert len(fetched_grants) == 1
+        assert fetched_grants[0] == declared_fqn
+
+
 class TestGrantsReportedUnderASynonym:
     """Snowflake reports a grant on an MCP server as CORTEX_AGENT_SERVER, while GRANT and
     CREATE call the object an MCP SERVER. Remote state and the manifest have to identify it