test: add test_anonymous_user to test_raft_service_levels

The primary goal of this test is to reproduce scylladb/scylladb#26040
so the fix (278019c328) can be backported
to older branches.

Scenario: connect via CQL as an anonymous user and verify that the
`sl:default` scheduling group is used. Before the fix for #26040
`main` scheduling group was incorrectly used instead of `sl:default`.

Control connections may legitimately use `sl:driver`, so the test
accepts those occurrences while still asserting that regular anonymous
queries use `sl:default`.

This adds explicit coverage on master. After scylladb#24411 was
implemented, some other tests started to fail when scylladb#26040
was unfixed. However, none of the tests asserted this exact behavior.

Refs: scylladb/scylladb#26040
Refs: scylladb/scylladb#26581

Closes scylladb/scylladb#26589

test/cluster/auth_cluster/test_raft_service_levels.py

@@ -8,7 +8,7 @@ import time
 import asyncio
 import logging
 from test.pylib.rest_client import get_host_api_address, read_barrier
-from test.pylib.util import unique_name, wait_for_cql_and_get_hosts
+from test.pylib.util import unique_name, wait_for_cql_and_get_hosts, wait_for
 from test.pylib.manager_client import ManagerClient
 from test.cluster.util import trigger_snapshot, wait_until_topology_upgrade_finishes, enter_recovery_state, reconnect_driver, \
         delete_raft_topology_state, delete_raft_data_and_upgrade_state, wait_until_upgrade_finishes, \
@@ -675,3 +675,29 @@ async def test_driver_service_level_used_for_driver_queries(manager: ManagerClie
     await cql.run_async(f"CREATE SERVICE LEVEL driver", host=h)
     cql, func = await get_control_connection_query_function(manager)
     await _verify_tasks_processed_metrics(manager, server, 'sl:driver', 'sl:test', func)
+
+# Reproduces scylladb/scylladb#26040
+@pytest.mark.asyncio
+async def test_anonymous_user(manager: ManagerClient) -> None:
+    allow_all_config = {'authenticator':'AllowAllAuthenticator', 'authorizer':'AllowAllAuthorizer'}
+    server = await manager.server_add(config=allow_all_config)
+    cql = manager.get_cql()
+    [h] = await wait_for_cql_and_get_hosts(cql, [server], time.time() + 60)
+
+    async def connections_ready():
+        rows = list(cql.execute("SELECT connection_stage, username, scheduling_group FROM system.clients"))
+        if len(rows) == 0:
+            return None
+        for row in rows:
+            if row.connection_stage != "READY":
+                return None
+        return rows
+
+    rows = await wait_for(connections_ready, time.time() + 60)
+    for r in rows:
+        assert r.username == 'anonymous'
+        assert r.scheduling_group in ['sl:default', 'sl:driver']
+        if r.scheduling_group == 'sl:default':
+            return
+
+    assert False, f"None of clients use sl:default, rows={rows}"