Fix change detection condition on KeeperMap read-only step

Unfortunately looking at `system.grants` to check whether specific grants have
been successfully revoked in ClickHouse is not perfect.

Let a user with a set of initial grants, including some ALTER statements. Doing
``REVOKE INSERT, ALTER UPDATE, ALTER DELETE ON <user> FROM <table>`` on this
user creates rows in ``system.grants`` for each of the grants with ``is_partial_revoke=1``.

If the user doesn't have ``ALTER`` statements in their initial grants, then
after executing ``REVOKE`` the ``system.grants`` table simply has no
corresponding rows.

Here's an example taken from running ``tests/integration/coordinator/plugins/clickhouse/test_steps.py::test_keeper_map_table_read_only_step``.

After ``REVOKE``:

```
SELECT *
FROM grants
WHERE
  user_name='bob' AND
  database='keeperdata' AND
  table='keepertable' AND
  access_type IN ('INSERT', 'ALTER UPDATE', 'ALTER DELETE')

user_name    role_name    access_type    database    table        column      is_partial_revoke    grant_option
-----------  -----------  -------------  ----------  -----------  --------  -------------------  --------------
bob                       INSERT         keeperdata  keepertable                              1               0
bob                       ALTER UPDATE   keeperdata  keepertable                              1               0
bob                       ALTER DELETE   keeperdata  keepertable                              1               0
```

After re-``GRANT``:

```
SELECT *
FROM grants
WHERE
  database='default' AND
  table='keeper_map'

user_name         role_name    access_type    database    table       column      is_partial_revoke    grant_option
----------------  -----------  -------------  ----------  ----------  --------  -------------------  --------------
aiven_monitoring               INSERT         default     keeper_map                              0               0
aiven_monitoring               ALTER UPDATE   default     keeper_map                              0               0
aiven_monitoring               ALTER DELETE   default     keeper_map                              0               0
alice                          INSERT         default     keeper_map                              0               0
alice                          ALTER UPDATE   default     keeper_map                              0               0
alice                          ALTER DELETE   default     keeper_map                              0               0
avnadmin                       INSERT         default     keeper_map                              1               1
avnadmin                       ALTER UPDATE   default     keeper_map                              1               1
avnadmin                       ALTER DELETE   default     keeper_map                              1               1
```

The fix is as follows: when checking for revoked DML grants, ignore rows in
system table that indicate partially revoked grants.

astacus/coordinator/plugins/clickhouse/steps.py

@@ -191,36 +191,40 @@ async def revoke_write_on_table(self, table: Table, user_name: bytes) -> None:
             f"REVOKE INSERT, ALTER UPDATE, ALTER DELETE ON {table.escaped_sql_identifier} FROM {escaped_user_name}"
         )
         await asyncio.gather(*(client.execute(revoke_statement.encode()) for client in self.clients))
-        await self.wait_for_access_type_grant(user_name=user_name, table=table, expected_count=0)
+        await self.wait_for_access_type_grant(user_name=user_name, table=table, expected_count=0, check_partial_revoke=True)
 
     async def grant_write_on_table(self, table: Table, user_name: bytes) -> None:
         escaped_user_name = escape_sql_identifier(user_name)
         grant_statement = (
             f"GRANT INSERT, ALTER UPDATE, ALTER DELETE ON {table.escaped_sql_identifier} TO {escaped_user_name}"
         )
         await asyncio.gather(*(client.execute(grant_statement.encode()) for client in self.clients))
-        await self.wait_for_access_type_grant(user_name=user_name, table=table, expected_count=3)
+        await self.wait_for_access_type_grant(user_name=user_name, table=table, expected_count=3, check_partial_revoke=False)
 
-    async def wait_for_access_type_grant(self, *, table: Table, user_name: bytes, expected_count: int) -> None:
+    async def wait_for_access_type_grant(
+        self, *, table: Table, user_name: bytes, expected_count: int, check_partial_revoke: bool
+    ) -> None:
         escaped_user_name = escape_sql_string(user_name)
         escaped_database = escape_sql_string(table.database)
         escaped_table = escape_sql_string(table.name)
 
-        async def check_function_count(client: ClickHouseClient) -> bool:
+        async def check_grant_count(client: ClickHouseClient) -> bool:
             statement = (
                 f"SELECT count() FROM grants "
                 f"WHERE user_name={escaped_user_name} "
                 f"AND database={escaped_database} "
                 f"AND table={escaped_table} "
                 f"AND access_type IN ('INSERT', 'ALTER UPDATE', 'ALTER DELETE')"
             )
+            if check_partial_revoke:
+                statement = statement + " AND is_partial_revoke = 0"
             num_grants_response = await client.execute(statement.encode())
             num_grants = int(cast(str, num_grants_response[0][0]))
             return num_grants == expected_count
 
         await wait_for_condition_on_every_node(
             clients=self.clients,
-            condition=check_function_count,
+            condition=check_grant_count,
             description="access grants changes to be enforced",
             timeout_seconds=60,
         )

tests/integration/coordinator/plugins/clickhouse/test_steps.py

@@ -109,6 +109,45 @@ class KeeperMapInfo(NamedTuple):
     user_client: ClickHouseClient
 
 
+def get_grant_database_usage_queries(user_name: str, database_names: Sequence[str]) -> list[bytes]:
+    database_privileges = [
+        "ALTER UPDATE",
+        "ALTER DELETE",
+        "ALTER COLUMN",
+        "ALTER MODIFY COMMENT",
+        "ALTER INDEX",
+        "ALTER PROJECTION",
+        "ALTER CONSTRAINT",
+        "ALTER TTL",
+        "ALTER SETTINGS",
+        "ALTER VIEW",
+        "CREATE TABLE",
+        "DROP TABLE",
+        "INSERT",
+        "OPTIMIZE",
+        "SELECT",
+        "SHOW",
+        "SYSTEM SYNC REPLICA",
+        "TRUNCATE",
+    ]
+    database_privileges_str = ", ".join(database_privileges)
+    return [
+        f"GRANT {database_privileges_str} ON {database_name}.* TO {user_name} WITH GRANT OPTION".encode()
+        for database_name in database_names
+    ]
+
+
+async def create_user_with_privileges(admin_client: ClickHouseClient, *, username: str, password: str) -> None:
+    await admin_client.execute(f"CREATE USER {username} IDENTIFIED WITH sha256_password BY '{password}'".encode())
+    response_rows = cast(
+        Sequence[tuple[str]],
+        await admin_client.execute(b"SELECT name FROM system.databases WHERE engine == 'Replicated' ORDER BY name"),
+    )
+    database_names = [row[0] for row in response_rows]
+    for query in get_grant_database_usage_queries(username, database_names):
+        await admin_client.execute(query)
+
+
 @pytest.fixture(name="keeper_table_context")
 async def fixture_keeper_table_context(
     ports: Ports, clickhouse_command: ClickHouseCommand, minio_bucket: MinioBucket
@@ -119,19 +158,20 @@ async def fixture_keeper_table_context(
     ):
         clickhouse = clickhouse_cluster.services[0]
         admin_client = get_clickhouse_client(clickhouse)
+        await admin_client.execute(
+            b"CREATE DATABASE `keeperdata` ENGINE = Replicated('/clickhouse/databases/keeperdata', '{my_shard}', '{my_replica}')"
+        )
         await setup_cluster_users([admin_client])
+        await create_user_with_privileges(admin_client, username="foobar", password="secret")
         for statement in [
-            b"CREATE DATABASE `keeperdata` ENGINE = Replicated('/clickhouse/databases/keeperdata', '{my_shard}', '{my_replica}')",
             b"CREATE TABLE `keeperdata`.`keepertable` (thekey UInt32, thevalue UInt32) ENGINE = KeeperMap('test', 1000) PRIMARY KEY thekey",
             b"INSERT INTO `keeperdata`.`keepertable` SELECT *, materialize(1) FROM numbers(3)",
-            b"CREATE USER bob IDENTIFIED WITH sha256_password BY 'secret'",
-            b"GRANT INSERT, SELECT, UPDATE, DELETE ON `keeperdata`.`keepertable` TO `bob`",
         ]:
             await admin_client.execute(statement)
         user_client = HttpClickHouseClient(
             host=clickhouse.host,
             port=clickhouse.port,
-            username="bob",
+            username="foobar",
             password="secret",
             timeout=10,
         )
@@ -142,7 +182,7 @@ async def fixture_keeper_table_context(
         yield KeeperMapInfo(context, admin_client, user_client)
 
 
-async def test_keeper_map_table_select_only_setting_modified(keeper_table_context: KeeperMapInfo) -> None:
+async def test_keeper_map_table_read_only_step(keeper_table_context: KeeperMapInfo) -> None:
     steps_context, admin_client, user_client = keeper_table_context
     read_only_step = KeeperMapTablesReadOnlyStep(clients=[admin_client], allow_writes=False)
     # After the read-only step, the user should only be able to select from the table