Wait until all users are created

Increase timeout time for individual tests. Do tests with Github runner + azure instance

Author: koletzilla

.github/workflows/test_cloud.yml

@@ -3,8 +3,7 @@ name: "test_cloud"
 
 on:  # yamllint disable-line rule:truthy
   pull_request:
-    branches:
-      - '*_cloud'
+    branches: main
   workflow_dispatch:
   schedule:
     - cron: '0 2 * * *'  # Nightly run at 2:00 AM UTC
@@ -13,32 +12,33 @@ jobs:
   cloud_smt_tests:
     name: ClickHouse Cloud SharedMergeTree Tests
     runs-on: ubuntu-latest
+    timeout-minutes: 90
 
     env:
       PYTHONPATH: dbt
-      DBT_CH_TEST_HOST: ${{ secrets.INTEGRATIONS_TEAM_TESTS_CLOUD_HOST_SMT }}
-      DBT_CH_TEST_PASSWORD: ${{ secrets.INTEGRATIONS_TEAM_TESTS_CLOUD_PASSWORD_SMT }}
+      DBT_CH_TEST_HOST: ${{ secrets.DBT_TESTS_CLOUD_HOST_SMT }}
+      DBT_CH_TEST_PASSWORD: ${{ secrets.DBT_TESTS_CLOUD_PASSWORD_SMT }}
       DBT_CH_TEST_CLUSTER_MODE: true
       DBT_CH_TEST_CLOUD: true
 
     steps:
       - name: Checkout
         uses: actions/checkout@v3
 
-      - name: Setup Python 3.11
+      - name: Setup Python 3.12
         uses: actions/setup-python@v4
         with:
-          python-version: '3.11'
+          python-version: '3.12'
 
       - name: Install requirements
         run: pip3 install . -r dev_requirements.txt
 
       - name: Run HTTP tests
         env:
           DBT_CH_TEST_PORT: 8443
-        run: pytest tests
+        run: pytest tests --timeout=240
 
       - name: Run Native tests
         env:
           DBT_CH_TEST_PORT: 9440
-        run: pytest tests
+        run: pytest tests --timeout=240

tests/integration/adapter/materialized_view/test_materialized_view.py

@@ -7,6 +7,7 @@
 
 import pytest
 from dbt.adapters.clickhouse.query import quote_identifier
+from dbt.tests.fixtures.project import TestProjInfo
 from dbt.tests.util import check_relation_types, run_dbt
 
 PEOPLE_SEED_CSV = """
@@ -193,14 +194,14 @@ def test_disabled_catchup(self, project):
 """
 
 
-def query_table_type(project, schema, table):
+def query_table_type(project: TestProjInfo, schema: str, table: str) -> str:
     table_type = project.run_sql(
         f"""
         select engine from system.tables where database = '{schema}' and name = '{table}'
     """,
         fetch="all",
     )
-    return table_type[0][0] if len(table_type) > 0 else None
+    return table_type[0][0] if len(table_type) > 0 else ''
 
 
 class TestUpdateMV:
@@ -300,7 +301,7 @@ def test_mv_is_dropped_on_full_refresh(self, project):
         # Step 4: Assert that target table is now a view and internal MV no longer exists
         assert query_table_type(project, schema_unquoted, 'hackers_mv') == "View"
         # Verify that the internal materialized view (_mv) no longer exists
-        assert query_table_type(project, schema_unquoted, 'hackers_mv_mv') is None
+        assert not query_table_type(project, schema_unquoted, 'hackers_mv_mv')
 
     def test_view_full_refresh_does_not_affect_existing_mv_with_mv_suffix(self, project):
         """

tests/integration/conftest.py

@@ -1,3 +1,4 @@
+import logging
 import os
 import random
 import sys
@@ -196,3 +197,48 @@ def wait_until_responsive(check, timeout, pause, clock=timeit.default_timer):
             return
         now = clock()
     raise Exception("Timeout reached while waiting on service!")
+
+
+def wait_for_users_in_cluster(client, cluster_name, users, timeout=30.0, pause=1.0):
+    start_time = timeit.default_timer()
+
+    result = client.query(f"""
+        SELECT count() as replica_count
+        FROM system.clusters
+        WHERE cluster = '{cluster_name}'
+    """)
+    replica_count = result.result_rows[0][0] if result.result_rows else 0
+
+    logging.debug(f"Found {replica_count} replicas in cluster '{cluster_name}'\n")
+
+    # Check if all users exist in all replicas
+    while (timeit.default_timer() - start_time) < timeout:
+        all_users_ready = True
+
+        user_list = ", ".join([f"'{user}'" for user in users])
+
+        user_check_query = f"""
+            SELECT name, count() as user_count
+            FROM clusterAllReplicas('{cluster_name}', system.users)
+            WHERE name IN ({user_list})
+            group by name
+        """
+        result = client.query(user_check_query)
+        for row in result.result_rows:
+            user_name, user_count = row
+            if user_count < replica_count:
+                logging.info(
+                    f"User '{user_name}' found in {user_count}/{replica_count} replicas. Waiting...\n"
+                )
+                all_users_ready = False
+                break
+
+        if all_users_ready:
+            logging.info(f"All {len(users)} users are created in all {replica_count} replicas\n")
+            return True
+
+        time.sleep(pause)
+
+    raise Exception(
+        f"Timeout reached: Not all users were created in all replicas within {timeout} seconds"
+    )