mirror of
https://github.com/hanzoai/datastore.git
synced 2026-09-26 13:55:07 +00:00
Alexey MilovidovandClaude Opus 4.8
295ced78d1
Follow-up to #105171, which added `ruff` with a deliberately narrow, bug-class-only rule set and left the unused-name and star-import rules off pending a dedicated cleanup pass. This enables that second batch in `pyproject.toml` and fixes every violation across the tree (the bulk in integration tests): - `F401` unused-import and `F841` unused-variable: removed, preserving any side-effecting right-hand side (e.g. `result = node.query(...)` becomes `node.query(...)`). Two `try: import ...` feature probes are kept with `# noqa: F401`. - `F541` f-string-missing-placeholders: dropped the redundant `f` prefix. - `F403`/`F405` star imports: every `from module import *` is expanded into explicit imports of the names actually used, taken from each name's real origin, so stdlib/third-party names import from their real source rather than through a test "convenience bundle". The now-empty `tests/integration/helpers/kafka/common_direct.py` bundle is removed and its three remaining consumers import `_VarintBytes`/`KafkaAdminClient` directly. Eliminating the star imports surfaced a real bug the wildcard had masked from `F821`: `generate_presets.py` in `test_prometheus_protocols` called an undefined `execute_instant_query_with_http_api`; fixed to the existing `execute_query_via_http_api`. No `black`/`isort` reformatting was applied (neither is enforced in the style check), so the diff is limited to the lint fixes themselves. Co-Authored-By: Claude Opus 4.8 (1M context) <[email protected]>
237 lines
7.4 KiB
Python
237 lines
7.4 KiB
Python
#!/usr/bin/env python3
|
|
|
|
import json
|
|
|
|
import pytest
|
|
import os
|
|
|
|
from multiprocessing.dummy import Pool
|
|
CONFIG_DIR = os.path.join(os.path.dirname(os.path.realpath(__file__)), "configs")
|
|
import helpers.keeper_utils as keeper_utils
|
|
from helpers.cluster import ClickHouseCluster
|
|
|
|
cluster = ClickHouseCluster(__file__)
|
|
node1 = cluster.add_instance(
|
|
"node1", main_configs=["configs/enable_keeper1.xml"], stay_alive=True
|
|
)
|
|
node2 = cluster.add_instance(
|
|
"node2", main_configs=["configs/enable_keeper2.xml"], stay_alive=True
|
|
)
|
|
node3 = cluster.add_instance(
|
|
"node3", main_configs=["configs/enable_keeper3.xml"], stay_alive=True
|
|
)
|
|
node4 = cluster.add_instance(
|
|
"node4", main_configs=["configs/enable_keeper4.xml"], stay_alive=True
|
|
)
|
|
node5 = cluster.add_instance(
|
|
"node5", main_configs=["configs/enable_keeper5.xml"], stay_alive=True
|
|
)
|
|
node6 = cluster.add_instance(
|
|
"node6", stay_alive=True
|
|
)
|
|
node7 = cluster.add_instance(
|
|
"node7", stay_alive=True
|
|
)
|
|
|
|
@pytest.fixture(scope="module")
|
|
def started_cluster():
|
|
try:
|
|
cluster.start()
|
|
|
|
yield cluster
|
|
|
|
finally:
|
|
cluster.shutdown()
|
|
|
|
def five_to_three_reconfig(started_cluster):
|
|
keeper_utils.wait_until_connected(cluster, node3)
|
|
zk = keeper_utils.get_fake_zk(cluster, "node3")
|
|
command = {
|
|
"max_action_wait_time_ms": 180000,
|
|
"max_total_wait_time_ms": 600000,
|
|
"preconditions": {
|
|
"leaders": [1, 2],
|
|
"members": [1, 2, 3, 4, 5]
|
|
},
|
|
"actions": [
|
|
{
|
|
"transfer_leadership": [3]
|
|
},
|
|
{
|
|
"remove_members": [1, 2]
|
|
},
|
|
{
|
|
"set_priority": [{"id": 4, "priority": 100}, {"id": 5, "priority": 100}]
|
|
},
|
|
{
|
|
"transfer_leadership": [4, 5]
|
|
},
|
|
{
|
|
"set_priority": [{"id": 3, "priority": 0}]
|
|
}
|
|
]
|
|
}
|
|
json_command = json.dumps(command)
|
|
print(json_command)
|
|
print(zk.get("/keeper/config"))
|
|
|
|
result_str = keeper_utils.send_4lw_cmd(started_cluster, node3, cmd="rcfg", port=9181, argument=json_command, timeout_sec=180)
|
|
print("Result:", result_str)
|
|
result = json.loads(result_str)
|
|
assert result["status"] == "ok"
|
|
|
|
leader = keeper_utils.get_leader(started_cluster, [node3, node4, node5])
|
|
assert leader.name in ["node4", "node5"]
|
|
|
|
content = zk.get("/keeper/config")[0].decode("utf-8")
|
|
assert 'server.1' not in content
|
|
assert 'server.2' not in content
|
|
|
|
assert 'server.3=node3:9234;participant;0' in content
|
|
assert 'server.4=node4:9234;participant;100' in content
|
|
assert 'server.5=node5:9234;participant;100' in content
|
|
|
|
|
|
def start(node):
|
|
node.start_clickhouse()
|
|
keeper_utils.wait_until_connected(cluster, node)
|
|
|
|
def three_to_five_reconfig(started_cluster):
|
|
keeper_utils.wait_until_connected(cluster, node3)
|
|
zk = keeper_utils.get_fake_zk(cluster, "node3")
|
|
|
|
node6.stop_clickhouse()
|
|
node6.copy_file_to_container(
|
|
os.path.join(CONFIG_DIR, "enable_keeper6.xml"),
|
|
"/etc/clickhouse-server/config.d/enable_keeper6.xml",
|
|
)
|
|
|
|
node7.stop_clickhouse()
|
|
node7.copy_file_to_container(
|
|
os.path.join(CONFIG_DIR, "enable_keeper7.xml"),
|
|
"/etc/clickhouse-server/config.d/enable_keeper7.xml",
|
|
)
|
|
|
|
p = Pool(3)
|
|
waiter1 = p.apply_async(start, (node6,))
|
|
waiter2 = p.apply_async(start, (node7,))
|
|
|
|
command = {
|
|
"max_action_wait_time_ms": 30000,
|
|
"max_total_wait_time_ms": 600000,
|
|
"preconditions": {
|
|
"leaders": [4, 5],
|
|
"members": [3, 4, 5]
|
|
},
|
|
"actions": [
|
|
{
|
|
"set_priority": [{"id": 3, "priority": 1}]
|
|
},
|
|
{
|
|
"add_members": [
|
|
{
|
|
"id": 6,
|
|
"endpoint": "node6:9234",
|
|
"priority": 100,
|
|
},
|
|
{
|
|
"id": 7,
|
|
"endpoint": "node7:9234",
|
|
"priority": 100,
|
|
},
|
|
],
|
|
"retry": 3
|
|
},
|
|
{
|
|
"transfer_leadership": [6, 7]
|
|
},
|
|
{
|
|
"set_priority": [{"id": 4, "priority": 0}, {"id": 5, "priority": 0}]
|
|
}
|
|
]
|
|
}
|
|
json_command = json.dumps(command)
|
|
print(json_command)
|
|
print(zk.get("/keeper/config"))
|
|
|
|
result_str = keeper_utils.send_4lw_cmd(started_cluster, node5, cmd="rcfg", port=9181, argument=json_command, timeout_sec=180)
|
|
print("Result:", result_str)
|
|
result = json.loads(result_str)
|
|
assert result["status"] == "ok"
|
|
waiter1.wait()
|
|
waiter2.wait()
|
|
|
|
leader = keeper_utils.get_leader(started_cluster, [node3, node4, node5, node6, node7])
|
|
assert leader.name in ["node6", "node7"]
|
|
|
|
content = zk.get("/keeper/config")[0].decode("utf-8")
|
|
assert 'server.1' not in content
|
|
assert 'server.2' not in content
|
|
assert 'server.3=node3:9234;participant;1' in content
|
|
assert 'server.4=node4:9234;participant;0' in content
|
|
assert 'server.5=node5:9234;participant;0' in content
|
|
assert 'server.6=node6:9234;participant;100' in content
|
|
assert 'server.7=node7:9234;participant;100' in content
|
|
|
|
|
|
def test_reconfig_option1(started_cluster):
|
|
keeper_utils.wait_until_connected(cluster, node3)
|
|
zk = keeper_utils.get_fake_zk(cluster, "node3")
|
|
zk.create("/test_reconfig_option1", b"data1")
|
|
content = zk.get("/keeper/config")[0].decode("utf-8")
|
|
if 'server.1' in content:
|
|
five_to_three_reconfig(started_cluster)
|
|
else:
|
|
three_to_five_reconfig(started_cluster)
|
|
|
|
assert zk.get("/test_reconfig_option1")[0] == b"data1"
|
|
|
|
def test_reconfig_option2(started_cluster):
|
|
keeper_utils.wait_until_connected(cluster, node3)
|
|
zk = keeper_utils.get_fake_zk(cluster, "node3")
|
|
zk.create("/test_reconfig_option2", b"data2")
|
|
content = zk.get("/keeper/config")[0].decode("utf-8")
|
|
if 'server.1' in content:
|
|
five_to_three_reconfig(started_cluster)
|
|
else:
|
|
three_to_five_reconfig(started_cluster)
|
|
|
|
assert zk.get("/test_reconfig_option2")[0] == b"data2"
|
|
|
|
def test_no_remove_itself(started_cluster):
|
|
command = {
|
|
"actions": [
|
|
{
|
|
"remove_members": [3]
|
|
},
|
|
]
|
|
}
|
|
json_command = json.dumps(command)
|
|
result_str = keeper_utils.send_4lw_cmd(started_cluster, node3, cmd="rcfg", port=9181, argument=json_command, timeout_sec=300)
|
|
result = json.loads(result_str)
|
|
assert result["status"] == "error"
|
|
assert "Reconfigure command cannot remove current server id" in result["message"]
|
|
|
|
def test_precondition_failure(started_cluster):
|
|
command = {
|
|
"preconditions": {
|
|
"leaders": [3],
|
|
},
|
|
}
|
|
json_command = json.dumps(command)
|
|
result_str = keeper_utils.send_4lw_cmd(started_cluster, node3, cmd="rcfg", port=9181, argument=json_command, timeout_sec=300)
|
|
result = json.loads(result_str)
|
|
assert result["status"] == "error"
|
|
assert "expected leader id" in result["message"]
|
|
|
|
command = {
|
|
"preconditions": {
|
|
"members": [5,6,7,8,9],
|
|
},
|
|
}
|
|
json_command = json.dumps(command)
|
|
result_str = keeper_utils.send_4lw_cmd(started_cluster, node3, cmd="rcfg", port=9181, argument=json_command, timeout_sec=300)
|
|
result = json.loads(result_str)
|
|
assert result["status"] == "error"
|
|
assert "found in cluster, but precondition" in result["message"]
|