Skip to content

Commit 79107eb

Browse files
Michicosunilejn
Michicosun
authored and
ilejn
committed
fix integration tests
Signed-off-by: Ilya Golshtein <igolshtein@altinity.com>
1 parent 792180a commit 79107eb

8 files changed

Lines changed: 166 additions & 173 deletions

File tree

‎tests/integration/helpers/blobs.py‎

Lines changed: 38 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,38 @@
1+
import time
2+
import typing
3+
import minio
4+
5+
6+
def list_blobs(minio_client: minio.Minio, bucket="root", path="data/"):
7+
objects = minio_client.list_objects(bucket, path, recursive=True)
8+
return [obj.object_name for obj in objects]
9+
10+
def wait_blobs_synchronization(minio_client: minio.Minio, expected_objects_list: typing.List[str], bucket="root", path="data/"):
11+
print(f"Waiting blobs in {bucket}/{path} will be equal to {expected_objects_list}")
12+
13+
for i in range(100):
14+
existing_objects = list_blobs(minio_client, bucket, path)
15+
print(f"[{i}] Existing S3 objects: {existing_objects}")
16+
17+
if set(existing_objects) == set(expected_objects_list):
18+
print("Blobs set validated")
19+
return
20+
21+
time.sleep(1)
22+
23+
assert False, f"Failed to wait blobs synchronization: {set(existing_objects)} != {set(expected_objects_list)}"
24+
25+
def wait_blobs_count_synchronization(minio_client: minio.Minio, expected_objects_count: int, bucket="root", path="data/"):
26+
print(f"Waiting count of blobs in {bucket}/{path} will be equal to {expected_objects_count}")
27+
28+
for i in range(100):
29+
existing_objects = list_blobs(minio_client, bucket, path)
30+
print(f"[{i}] Existing S3 objects: {existing_objects}")
31+
32+
if len(existing_objects) == expected_objects_count:
33+
print("Blobs set validated")
34+
return
35+
36+
time.sleep(1)
37+
38+
assert False, f"Failed to wait blobs synchronization: {len(existing_objects)} != {expected_objects_count}"

‎tests/integration/test_alter_moving_garbage/test.py‎

Lines changed: 1 addition & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -232,7 +232,7 @@ def test_delete_race_leftovers(cluster):
232232
# and it can be race condition between removing from remote_data_paths and deleting blobs
233233
all_remote_paths = set()
234234
known_remote_paths = set()
235-
for i in range(3):
235+
for i in range(100):
236236
known_remote_paths = set(
237237
node.query(
238238
f"SELECT remote_path FROM system.remote_data_paths WHERE disk_name = 's32'"

‎tests/integration/test_disk_configuration/test.py‎

Lines changed: 61 additions & 51 deletions
Original file line numberDiff line numberDiff line change
@@ -2,15 +2,12 @@
22

33
import pytest
44

5-
from helpers.client import QueryRuntimeException
65
from helpers.cluster import ClickHouseCluster
76
from helpers.config_cluster import minio_secret_key
8-
from minio.deleteobjects import DeleteObject
7+
from helpers.blobs import wait_blobs_count_synchronization
98

109
cluster = ClickHouseCluster(__file__)
1110

12-
TABLE_NAME = "test"
13-
1411

1512
@pytest.fixture(scope="module")
1613
def start_cluster():
@@ -61,39 +58,23 @@ def start_cluster():
6158
cluster.shutdown()
6259

6360

64-
def remove_minio_objects(minio, path: str):
65-
retry_count = 3
66-
remaining_files = map(
67-
lambda x: DeleteObject(x.object_name),
68-
minio.list_objects(cluster.minio_bucket, path, recursive=True),
69-
)
70-
while len(list(remaining_files)) > 0 and retry_count > 0:
71-
errors = minio.remove_objects(cluster.minio_bucket, remaining_files)
72-
for error in errors:
73-
logging.error(f"error occurred when deleting minio object: {error}")
74-
75-
remaining_files = map(
76-
lambda x: DeleteObject(x.object_name),
77-
minio.list_objects(cluster.minio_bucket, path, recursive=True),
78-
)
79-
retry_count -= 1
80-
assert len(list(remaining_files)) == 0, remaining_files
81-
82-
8361
def test_merge_tree_disk_setting(start_cluster):
62+
TABLE_NAME = "test_merge_tree_disk_setting"
8463
node1 = cluster.instances["node1"]
64+
minio = cluster.minio_client
65+
66+
wait_blobs_count_synchronization(minio, 0, bucket="root", path="data")
67+
wait_blobs_count_synchronization(minio, 0, bucket="root", path="data2")
8568

8669
node1.query(
8770
f"""
88-
DROP TABLE IF EXISTS {TABLE_NAME};
8971
CREATE TABLE {TABLE_NAME} (a Int32)
9072
ENGINE = MergeTree()
9173
ORDER BY tuple()
9274
SETTINGS disk = 's3';
9375
"""
9476
)
9577

96-
minio = cluster.minio_client
9778
count = len(list(minio.list_objects(cluster.minio_bucket, "data/", recursive=True)))
9879

9980
node1.query(f"INSERT INTO {TABLE_NAME} SELECT number FROM numbers(100)")
@@ -105,7 +86,6 @@ def test_merge_tree_disk_setting(start_cluster):
10586

10687
node1.query(
10788
f"""
108-
DROP TABLE IF EXISTS {TABLE_NAME}_2;
10989
CREATE TABLE {TABLE_NAME}_2 (a Int32)
11090
ENGINE = MergeTree()
11191
ORDER BY tuple()
@@ -159,15 +139,20 @@ def test_merge_tree_disk_setting(start_cluster):
159139
node1.query(f"DROP TABLE {TABLE_NAME} SYNC")
160140
node1.query(f"DROP TABLE {TABLE_NAME}_2 SYNC")
161141

162-
remove_minio_objects(minio, "data/")
142+
wait_blobs_count_synchronization(minio, 0, bucket="root", path="data")
143+
wait_blobs_count_synchronization(minio, 0, bucket="root", path="data2")
163144

164145

165146
def test_merge_tree_custom_disk_setting(start_cluster):
147+
TABLE_NAME = "test_merge_tree_custom_disk_setting"
166148
node1 = cluster.instances["node1"]
167149
node2 = cluster.instances["node2"]
168-
150+
minio = cluster.minio_client
169151
zk_client = start_cluster.get_kazoo_client("zoo1")
170152

153+
wait_blobs_count_synchronization(minio, 0, bucket="root", path="data")
154+
wait_blobs_count_synchronization(minio, 0, bucket="root", path="data2")
155+
171156
if not zk_client.exists("/minio"):
172157
zk_client.create("/minio")
173158
if not zk_client.exists("/minio/access_key_id"):
@@ -176,7 +161,6 @@ def test_merge_tree_custom_disk_setting(start_cluster):
176161
zk_client.set("/minio/access_key_id", b"minio")
177162
node1.query(
178163
f"""
179-
DROP TABLE IF EXISTS {TABLE_NAME};
180164
CREATE TABLE {TABLE_NAME} (a Int32)
181165
ENGINE = MergeTree()
182166
ORDER BY tuple()
@@ -191,7 +175,6 @@ def test_merge_tree_custom_disk_setting(start_cluster):
191175

192176
# Check that data was indeed created on s3 with the needed path in s3
193177

194-
minio = cluster.minio_client
195178
count = len(list(minio.list_objects(cluster.minio_bucket, "data/", recursive=True)))
196179

197180
node1.query(f"INSERT INTO {TABLE_NAME} SELECT number FROM numbers(100)")
@@ -205,7 +188,6 @@ def test_merge_tree_custom_disk_setting(start_cluster):
205188

206189
node1.query(
207190
f"""
208-
DROP TABLE IF EXISTS {TABLE_NAME}_2;
209191
CREATE TABLE {TABLE_NAME}_2 (a Int32)
210192
ENGINE = MergeTree()
211193
ORDER BY tuple()
@@ -228,11 +210,8 @@ def test_merge_tree_custom_disk_setting(start_cluster):
228210

229211
# Check that data for a disk with a different path was created on the different path
230212

231-
remove_minio_objects(minio, "data2/")
232-
233213
node1.query(
234214
f"""
235-
DROP TABLE IF EXISTS {TABLE_NAME}_3;
236215
CREATE TABLE {TABLE_NAME}_3 (a Int32)
237216
ENGINE = MergeTree()
238217
ORDER BY tuple()
@@ -292,7 +271,6 @@ def test_merge_tree_custom_disk_setting(start_cluster):
292271
replica = "{replica}"
293272
node1.query(
294273
f"""
295-
DROP TABLE IF EXISTS {TABLE_NAME}_4;
296274
CREATE TABLE {TABLE_NAME}_4 ON CLUSTER 'cluster' (a Int32)
297275
ENGINE=ReplicatedMergeTree('/clickhouse/tables/tbl/', '{replica}')
298276
ORDER BY tuple()
@@ -345,16 +323,20 @@ def test_merge_tree_custom_disk_setting(start_cluster):
345323
node1.query(f"DROP TABLE {TABLE_NAME}_4 SYNC")
346324
node2.query(f"DROP TABLE {TABLE_NAME}_4 SYNC")
347325

326+
wait_blobs_count_synchronization(minio, 0, bucket="root", path="data")
327+
wait_blobs_count_synchronization(minio, 0, bucket="root", path="data2")
328+
348329

349330
def test_merge_tree_nested_custom_disk_setting(start_cluster):
331+
TABLE_NAME = "test_merge_tree_nested_custom_disk_setting"
350332
node = cluster.instances["node1"]
351-
352333
minio = cluster.minio_client
353-
remove_minio_objects(minio, "data/")
334+
335+
wait_blobs_count_synchronization(minio, 0, bucket="root", path="data")
336+
wait_blobs_count_synchronization(minio, 0, bucket="root", path="data2")
354337

355338
node.query(
356339
f"""
357-
DROP TABLE IF EXISTS {TABLE_NAME} SYNC;
358340
CREATE TABLE {TABLE_NAME} (a Int32)
359341
ENGINE = MergeTree() order by tuple()
360342
SETTINGS disk = disk(
@@ -393,14 +375,22 @@ def test_merge_tree_nested_custom_disk_setting(start_cluster):
393375
assert expected.strip() in node.query(f"SHOW CREATE TABLE {TABLE_NAME}").strip()
394376
node.query(f"DROP TABLE {TABLE_NAME} SYNC")
395377

378+
wait_blobs_count_synchronization(minio, 0, bucket="root", path="data")
379+
wait_blobs_count_synchronization(minio, 0, bucket="root", path="data2")
380+
396381

397382
def test_merge_tree_setting_override(start_cluster):
383+
TABLE_NAME = "test_merge_tree_setting_override"
398384
node = cluster.instances["node3"]
385+
minio = cluster.minio_client
386+
387+
wait_blobs_count_synchronization(minio, 0, bucket="root", path="data")
388+
wait_blobs_count_synchronization(minio, 0, bucket="root", path="data2")
389+
399390
assert (
400391
"MergeTree settings `storage_policy` and `disk` cannot be specified at the same time"
401392
in node.query_and_get_error(
402393
f"""
403-
DROP TABLE IF EXISTS {TABLE_NAME};
404394
CREATE TABLE {TABLE_NAME} (a Int32)
405395
ENGINE = MergeTree()
406396
ORDER BY tuple()
@@ -413,7 +403,6 @@ def test_merge_tree_setting_override(start_cluster):
413403
"MergeTree settings `storage_policy` and `disk` cannot be specified at the same time"
414404
in node.query_and_get_error(
415405
f"""
416-
DROP TABLE IF EXISTS {TABLE_NAME} SYNC;
417406
CREATE TABLE {TABLE_NAME} (a Int32)
418407
ENGINE = MergeTree()
419408
ORDER BY tuple()
@@ -423,11 +412,14 @@ def test_merge_tree_setting_override(start_cluster):
423412
)
424413
)
425414

415+
node.query(f"DROP TABLE {TABLE_NAME} SYNC")
416+
wait_blobs_count_synchronization(minio, 0, bucket="root", path="data")
417+
wait_blobs_count_synchronization(minio, 0, bucket="root", path="data2")
418+
426419
assert (
427420
"MergeTree settings `storage_policy` and `disk` cannot be specified at the same time"
428421
in node.query_and_get_error(
429422
f"""
430-
DROP TABLE IF EXISTS {TABLE_NAME} SYNC;
431423
CREATE TABLE {TABLE_NAME} (a Int32)
432424
ENGINE = MergeTree()
433425
ORDER BY tuple()
@@ -437,11 +429,14 @@ def test_merge_tree_setting_override(start_cluster):
437429
)
438430
)
439431

432+
node.query(f"DROP TABLE {TABLE_NAME} SYNC")
433+
wait_blobs_count_synchronization(minio, 0, bucket="root", path="data")
434+
wait_blobs_count_synchronization(minio, 0, bucket="root", path="data2")
435+
440436
assert (
441437
"New storage policy `local` shall contain volumes of the old storage policy `s3`"
442438
in node.query_and_get_error(
443439
f"""
444-
DROP TABLE IF EXISTS {TABLE_NAME};
445440
CREATE TABLE {TABLE_NAME} (a Int32)
446441
ENGINE = MergeTree()
447442
ORDER BY tuple()
@@ -451,12 +446,15 @@ def test_merge_tree_setting_override(start_cluster):
451446
)
452447
)
453448

449+
node.query(f"DROP TABLE {TABLE_NAME} SYNC")
450+
wait_blobs_count_synchronization(minio, 0, bucket="root", path="data")
451+
wait_blobs_count_synchronization(minio, 0, bucket="root", path="data2")
452+
454453
# Using default policy so storage_policy and disk are not set at the same time
455454
assert (
456455
"New storage policy `__disk_local` shall contain disks of the old storage policy `hybrid`"
457456
in node.query_and_get_error(
458457
f"""
459-
DROP TABLE IF EXISTS {TABLE_NAME};
460458
CREATE TABLE {TABLE_NAME} (a Int32)
461459
ENGINE = MergeTree()
462460
ORDER BY tuple();
@@ -465,9 +463,12 @@ def test_merge_tree_setting_override(start_cluster):
465463
)
466464
)
467465

466+
node.query(f"DROP TABLE {TABLE_NAME} SYNC")
467+
wait_blobs_count_synchronization(minio, 0, bucket="root", path="data")
468+
wait_blobs_count_synchronization(minio, 0, bucket="root", path="data2")
469+
468470
assert "Unknown storage policy" in node.query_and_get_error(
469471
f"""
470-
DROP TABLE IF EXISTS {TABLE_NAME};
471472
CREATE TABLE {TABLE_NAME} (a Int32)
472473
ENGINE = MergeTree()
473474
ORDER BY tuple()
@@ -477,7 +478,6 @@ def test_merge_tree_setting_override(start_cluster):
477478

478479
assert "Unknown disk" in node.query_and_get_error(
479480
f"""
480-
DROP TABLE IF EXISTS {TABLE_NAME};
481481
CREATE TABLE {TABLE_NAME} (a Int32)
482482
ENGINE = MergeTree()
483483
ORDER BY tuple()
@@ -487,7 +487,6 @@ def test_merge_tree_setting_override(start_cluster):
487487

488488
node.query(
489489
f"""
490-
DROP TABLE IF EXISTS {TABLE_NAME} SYNC;
491490
CREATE TABLE {TABLE_NAME} (a Int32)
492491
ENGINE = MergeTree()
493492
ORDER BY tuple()
@@ -500,16 +499,18 @@ def test_merge_tree_setting_override(start_cluster):
500499
"""
501500
)
502501

503-
minio = cluster.minio_client
504502
node.query(f"INSERT INTO {TABLE_NAME} SELECT number FROM numbers(100)")
505503
assert int(node.query(f"SELECT count() FROM {TABLE_NAME}")) == 100
506504
assert (
507505
len(list(minio.list_objects(cluster.minio_bucket, "data/", recursive=True))) > 0
508506
)
507+
node.query(f"DROP TABLE {TABLE_NAME} SYNC")
508+
509+
wait_blobs_count_synchronization(minio, 0, bucket="root", path="data")
510+
wait_blobs_count_synchronization(minio, 0, bucket="root", path="data2")
509511

510512
node.query(
511513
f"""
512-
DROP TABLE IF EXISTS {TABLE_NAME} SYNC;
513514
CREATE TABLE {TABLE_NAME} (a Int32)
514515
ENGINE = MergeTree()
515516
ORDER BY tuple()
@@ -525,6 +526,10 @@ def test_merge_tree_setting_override(start_cluster):
525526
)
526527
node.query(f"DROP TABLE {TABLE_NAME} SYNC")
527528

529+
wait_blobs_count_synchronization(minio, 0, bucket="root", path="data")
530+
wait_blobs_count_synchronization(minio, 0, bucket="root", path="data2")
531+
532+
528533
@pytest.mark.parametrize("use_node", ["node1", "node2"])
529534
def test_merge_tree_custom_encrypted_disk_include(start_cluster, use_node):
530535
"""Test that encrypted disk configuration works with include parameter.
@@ -534,11 +539,15 @@ def test_merge_tree_custom_encrypted_disk_include(start_cluster, use_node):
534539
- The include mechanism correctly merges encryption keys into the disk config
535540
- Data can be successfully encrypted and decrypted using the included keys
536541
"""
542+
TABLE_NAME = f"test_merge_tree_custom_encrypted_disk_include_{use_node}"
537543
node = cluster.instances[use_node]
544+
minio = cluster.minio_client
545+
546+
wait_blobs_count_synchronization(minio, 0, bucket="root", path="data")
547+
wait_blobs_count_synchronization(minio, 0, bucket="root", path="data2")
538548

539549
node.query(
540550
f"""
541-
DROP TABLE IF EXISTS {TABLE_NAME}_encrypted;
542551
CREATE TABLE {TABLE_NAME}_encrypted (
543552
id Int32,
544553
data String
@@ -568,9 +577,10 @@ def test_merge_tree_custom_encrypted_disk_include(start_cluster, use_node):
568577
result = node.query(f"SELECT data FROM {TABLE_NAME}_encrypted WHERE id = 1")
569578
assert result.strip() == "test_data"
570579

571-
minio = cluster.minio_client
572580
s3_objects = list(minio.list_objects(cluster.minio_bucket, "data/", recursive=True))
573581
assert len(s3_objects) > 0, "Data should be written to S3"
574582

575583
node.query(f"DROP TABLE {TABLE_NAME}_encrypted SYNC")
576-
remove_minio_objects(minio, "data/")
584+
585+
wait_blobs_count_synchronization(minio, 0, bucket="root", path="data")
586+
wait_blobs_count_synchronization(minio, 0, bucket="root", path="data2")

0 commit comments

Comments
 (0)