mirror of
https://github.com/milvus-io/milvus.git
synced 2026-10-28 17:05:53 +08:00
f929d68ba6
issue: #52114 Makes `ruff check` and `ruff format --check` pass over all of `tests/`, so the `Ruff (changed files only)` job stops failing PRs for violations they did not introduce (most recently #51903, which got 69 errors from an 8-line change). Almost all of this is mechanical. The behavioural changes are called out individually below: one `NameError` fix, three shadowed duplicate definitions, 13 assertion gaps, and two undefined names that only became visible once star imports were expanded. ### Layout Seven reviewable commits: | commit | what | files | |---|---|---| | 1 | `ruff format` | 170 | | 2 | `ruff check --fix` (safe fixes only, no `--unsafe-fixes`) | 156 | | 3 | F821 fix + `tests/ruff.toml` | 4 | | 4 | three shadowed duplicate definitions | 4 | | 5 | close 13 assertion gaps behind F841 | 7 | | 6 | expand star imports, re-enable F403/F405/F821 | 33 | | 7 | enable E402, E712, E722 | 16 | 1745 violations → 0. Only `tests/` is touched. ### The undefined-name bug `chaos/test_chaos_memory_stress.py`: `TestMemoryStressReplica.prepare_collection` referenced bare `nb` and `dim`. Those are class attributes, never module-level names, so the fixture raised `NameError` the moment it ran — and the line directly above it already used `self.dim`. Changed to `self.nb` / `self.dim`. This was the only F821 in `tests/`; **the rule stays enabled**. ### Three shadowed duplicate definitions Three places define the same name twice in the same scope. Python keeps the later one, so the earlier definition is dead while still reading as live code — a test that looks like it exists and never runs. **`restful_client/testcases/test_vector_operations.py` — `TestDeleteVector`** ```python @pytest.mark.L0 class TestDeleteVector(TestBase): # line 778 — shadowed, never ran def test_delete_vector_default(...) # 2x2 parametrized @pytest.mark.L1 class TestDeleteVector(TestBase): # line 838 — wins def test_delete_vector_with_invalid_api_key(...) def test_delete_vector_with_invalid_collection_name(...) def test_delete_vector_with_non_primary_key(...) ``` The second holds only negative cases, and the file already uses `TestInsertVectorNegative` / `TestSearchVectorNegative`, so this is a copy whose class name was never updated. Renamed it to `TestDeleteVectorNegative`. **This is the one to look at:** it brings back an `L0` case, so 4 parametrizations of `test_delete_vector_default` start running in PR CI. They have never executed, so if the restful delete path has drifted since they were written, this is where it surfaces. **`python_client/testcases/test_compaction.py` — `test_compact_after_delete`** Defined twice in `TestCompactionParams`, covering different things: the shadowed one (L3, `@parametrize("delete_pos", [1, tmp_nb // 2])`) covers *partial* deletes plus a compaction-plan check; the surviving one (L2) deletes everything. There is already a separate `test_compact_after_delete_all`, so partial-delete coverage existed only in the shadowed case and was not running. Renamed it to `test_compact_after_delete_partial`; the surviving definition is untouched. **`python_client/utils/util_pymilvus.py` — `add_vector_field`** Defined twice with different signatures: ```python def add_vector_field(entities, is_normal=False): # line 455 — shadowed ... entities.append(field) return entities def add_vector_field(nb, dimension=default_dim): # line 527 — wins field_name = gen_unique_str() field = {...} # built, then discarded return field_name # returns only the name ``` Removed the second — both the broken one and the one that was winning. Grepped the whole repo: **neither definition has a caller**; every `add_vector_field` hit is something else (`test_milvus_client_add_vector_field` test methods, the `Op.add_vector_field` enum member in `chaos/checker.py`, a `Checker` method of the same name). Dead-code removal, not an API change. **Net effect on what executes:** `test_delete_vector_default` ×4 (**L0**) and `test_compact_after_delete_partial` ×2 (L3) start running; nothing stops running; the 3 negative cases keep running under the new class name. Computed by parsing each file before and after and resolving shadowing the way Python does, not by reading the diff. F811 is now clean across `tests/` with no per-file suppressions. ### Star imports expanded, F403/F405/F821 re-enabled 33 of 307 files used `from utils.util_pymilvus import *` and friends. Inside those, pyflakes cannot tell a typo from a star-imported name — it downgrades everything unresolved to F405, and **F821 never fires**. The `NameError` fixed in commit 3 was only catchable because that file happens not to use a star import. The 379 F405 hits were only **58 distinct names**. All are now explicit, resolved by walking the star modules in source order (later import wins, matching Python). Names that `util_pymilvus` merely re-exports come from their real source now — `DataType` from `pymilvus`, `np` from `numpy`, `log` from `utils.util_log` — rather than being laundered through it. Turning F821 on immediately caught **two real undefined names**, both in `rate_limit/test_rate_limit.py` (L3, so it has never run in CI): ```python for collection_num in range(collectionRateLimit): # no such name # ^ the parametrize arg is `collection_rate_limit`, and it is # a string ("10"/"500"), so range() needs int(). # The same test uses the correct name 30 lines below. mil.uninstall(release_name, namespace=NAMESPACE) # no such name # ^ MilvusOperator.uninstall has exactly this signature, and other suites do the # same under `mic`/`milvus_op` — the instantiation was just missing. ``` Both fixed. `test_milvus_client_data_integrity_new.py` needed 15 `noqa` markers instead: its `prepare_data` fixture injects those names via `globals()[field] = ...`, which no static analysis can follow. That block has other problems — a `float is None` check against the builtin, an always-true `if "is null" or "IS NULL" in expression`, leftover `binbin_debug1` logging — but rewriting fixture logic does not belong here; noted in #52116. ### 13 assertion gaps closed Of ~80 F841 hits, most are harmless scratch values. 13 were a call whose result was bound and then never checked. Highlights: - `test_milvus_client_database.py`: two tests apply a non-existent property key (`data.replica.number`, note `data.` not `database.`), describe the database, and drop it without checking anything. Their docstrings say `actual: run successfully` / `nothing changed` — the intent was to prove nothing changed. Now asserted. - `test_milvus_client_alias.py`: two assertions were commented out, **and they were commented out for a reason** — `list_aliases()[0]` returns a dict, so `assert alias in aliases` would have tested dict keys. Rewritten against `[0]["aliases"]` (how the rest of the file does it) and re-enabled. In the skipped `alias_default` test the calls were also missing the required `collection_name` argument. The remaining ~65 F841 stay, and the rule stays off. ### E402, E712, E722 also enabled **E722** (37 sites): mechanical `except:` → `except Exception:`. Checked the two ways this could change behaviour, both empty: no bare-except try-block can raise `SystemExit`/`KeyboardInterrupt`, and **none calls `pytest.skip`/`fail`/`xfail`**. That second one matters — pytest's `Skipped` derives from `BaseException`, so a bare except silently eats it while `except Exception` would not. 0 of 37, so behaviour is unchanged. **E402** (5 sites): imports that were just in the wrong place — `import re` sat 70 lines into `check/param_check.py`, and the two `resource_group` modules imported `rich` after their module-level constants. Nothing to do with `sys.path`, which is what an earlier version of this description wrongly claimed. **E712** (5 sites), gone through one at a time rather than bulk-rewritten: | site | value | change | |---|---|---| | `test_milvus_client_partition.py` | `has_partition()` | → `is False` — stricter than `not result`, which would pass on `None` | | `test_restart.py` | JSON status flag | → `is True` | | `test_restart.py` | local `break_flag` | → plain truth test, it is a real bool | | `test_query.py` | wrapper `check_result` | → `is True` | | `test_milvus_client_ttl.py` | `hit.get(bool_field)` | **kept as `== False` + noqa** | That last one is the reason this rule cannot be bulk-fixed: `.get()` returns `None` for a missing field, so `not ...` would pass exactly where the assertion should fail, and `is False` would break on a numpy bool. ### What ruff.toml still disables | rule | n | reason | |---|---|---| | F841 | ~65 | tests deliberately keep call results; the 13 that were genuine assertion gaps are fixed above | | UP031 | 95 | pure style rewrite that catches no bugs, and 27 sites use `% (...)` tuples where it is not equivalent | Everything else pyflakes/pycodestyle checks is on, including `F401`, `F811` and `F821`, so new code cannot introduce dead, duplicate or undefined names. Tracked in #52116. ### Verification Not just "ruff is green" — the mechanical passes were checked for semantic drift: - **Format pass:** parsed all 307 files before and after and compared ASTs with string constants normalized away → **zero structural differences**. The only AST-visible deltas are 146 string-constant changes, every one whitespace-only (docstring indentation / trailing whitespace); zero non-whitespace string changes. - **Import removals:** 142 imports were dropped. For each, checked whether the bound name still appears anywhere in its file. 115 have no remaining reference at all. The 27 that still matched textually were each read by hand and are all comments (`# pdb.set_trace()`, the commented-out checker dict in `test_load_with_checker.py`), docstring prose (`"input random collection name"`, `"Tuple[List[Dict]]"`), string literals (`"Collection ID"`, the `"config"` dict key), a different binding (`from time import sleep`), or attribute access through another module (`ct.CaseLabel`, `np.random`). **No real reference was removed.** - **Python version:** the pyupgrade rewrites emit `datetime.UTC` (3.11+) and `X | None` (3.10+). `tests/.python-version` is 3.12, `tests/scripts/python-env.sh` defaults to 3.12, and `tests/docker/Dockerfile` is `python:3.12-bullseye`. - **`compileall`** over all 307 files under Python 3.12 passes. - No external reference to `TestDeleteVector`, `test_compact_after_delete`, or `add_vector_field` exists anywhere in the repo (`*.py`, `*.sh`, `*.yaml`, `*.json`). ### Conflict surface Of the 100 currently-open PRs, 9 touch `tests/*.py` and only **2** overlap with this change: #51903 (3 of its 5 files) and #51856 (1 file). The other 7 touch files that were already format-clean. Worth merging promptly to keep that number small. `.pre-commit-config.yaml` needs no change — it already runs with `--config=tests/ruff.toml`, so local hooks and CI stay in sync. --------- Signed-off-by: xiaofanluan <xf@hjjaq.com> Co-authored-by: xiaofanluan <xf@hjjaq.com> Co-authored-by: Claude Opus 5 (1M context) <noreply@anthropic.com>
784 lines
34 KiB
Python
784 lines
34 KiB
Python
import sys
|
|
import time
|
|
from datetime import datetime
|
|
|
|
from pymilvus import utility
|
|
|
|
sys.path.append("..")
|
|
from check.func_check import ResponseChecker
|
|
from pymilvus import BulkInsertState
|
|
from pymilvus.orm.role import Role
|
|
from utils.api_request import api_request
|
|
from utils.util_log import test_log as log
|
|
|
|
TIMEOUT = 20
|
|
|
|
|
|
class ApiUtilityWrapper:
|
|
"""Method of encapsulating utility files"""
|
|
|
|
ut = utility
|
|
role = None
|
|
|
|
def do_bulk_insert(
|
|
self,
|
|
collection_name,
|
|
files="",
|
|
partition_name=None,
|
|
timeout=None,
|
|
using="default",
|
|
check_task=None,
|
|
check_items=None,
|
|
**kwargs,
|
|
):
|
|
working_tasks = self.get_bulk_insert_working_list()
|
|
log.info(f"before bulk load, there are {len(working_tasks)} working tasks")
|
|
log.info(f"files to load: {files}")
|
|
func_name = sys._getframe().f_code.co_name
|
|
res, is_succ = api_request(
|
|
[self.ut.do_bulk_insert, collection_name, files, partition_name, timeout, using], **kwargs
|
|
)
|
|
check_result = ResponseChecker(
|
|
res, func_name, check_task, check_items, is_succ, collection_name=collection_name, using=using
|
|
).run()
|
|
time.sleep(1)
|
|
working_tasks = self.get_bulk_insert_working_list()
|
|
log.info(f"after bulk load, there are {len(working_tasks)} working tasks")
|
|
return res, check_result
|
|
|
|
def get_bulk_insert_state(
|
|
self, task_id, timeout=None, using="default", check_task=None, check_items=None, **kwargs
|
|
):
|
|
func_name = sys._getframe().f_code.co_name
|
|
res, is_succ = api_request([self.ut.get_bulk_insert_state, task_id, timeout, using], **kwargs)
|
|
check_result = ResponseChecker(
|
|
res, func_name, check_task, check_items, is_succ, task_id=task_id, using=using
|
|
).run()
|
|
return res, check_result
|
|
|
|
def list_bulk_insert_tasks(
|
|
self, limit=0, collection_name=None, timeout=None, using="default", check_task=None, check_items=None, **kwargs
|
|
):
|
|
func_name = sys._getframe().f_code.co_name
|
|
res, is_succ = api_request([self.ut.list_bulk_insert_tasks, limit, collection_name, timeout, using], **kwargs)
|
|
check_result = ResponseChecker(
|
|
res, func_name, check_task, check_items, is_succ, limit=limit, collection_name=collection_name, using=using
|
|
).run()
|
|
return res, check_result
|
|
|
|
def get_bulk_insert_pending_list(self):
|
|
tasks = {}
|
|
for task in self.ut.list_bulk_insert_tasks():
|
|
if task.state == BulkInsertState.ImportPending:
|
|
tasks[task.task_id] = task
|
|
return tasks
|
|
|
|
def get_bulk_insert_working_list(self):
|
|
tasks = {}
|
|
for task in self.ut.list_bulk_insert_tasks():
|
|
if task.state in [BulkInsertState.ImportStarted]:
|
|
tasks[task.task_id] = task
|
|
return tasks
|
|
|
|
def list_all_bulk_insert_tasks(self, limit=0):
|
|
tasks, _ = self.list_bulk_insert_tasks(limit=limit)
|
|
pending = 0
|
|
started = 0
|
|
persisted = 0
|
|
completed = 0
|
|
failed = 0
|
|
failed_and_cleaned = 0
|
|
unknown = 0
|
|
for task in tasks:
|
|
print(task)
|
|
if task.state == BulkInsertState.ImportPending:
|
|
pending = pending + 1
|
|
elif task.state == BulkInsertState.ImportStarted:
|
|
started = started + 1
|
|
elif task.state == BulkInsertState.ImportPersisted:
|
|
persisted = persisted + 1
|
|
elif task.state == BulkInsertState.ImportCompleted:
|
|
completed = completed + 1
|
|
elif task.state == BulkInsertState.ImportFailed:
|
|
failed = failed + 1
|
|
elif task.state == BulkInsertState.ImportFailedAndCleaned:
|
|
failed_and_cleaned = failed_and_cleaned + 1
|
|
else:
|
|
unknown = unknown + 1
|
|
|
|
log.info(
|
|
"There are",
|
|
len(tasks),
|
|
"bulkload tasks.",
|
|
pending,
|
|
"pending,",
|
|
started,
|
|
"started,",
|
|
persisted,
|
|
"persisted,",
|
|
completed,
|
|
"completed,",
|
|
failed,
|
|
"failed",
|
|
failed_and_cleaned,
|
|
"failed_and_cleaned",
|
|
unknown,
|
|
"unknown",
|
|
)
|
|
|
|
def wait_for_bulk_insert_tasks_completed(
|
|
self, task_ids, target_state=BulkInsertState.ImportCompleted, timeout=None, using="default", **kwargs
|
|
):
|
|
tasks_state_distribution = {"success": set(), "failed": set(), "in_progress": set()}
|
|
tasks_state = {}
|
|
if timeout is not None:
|
|
task_timeout = timeout
|
|
else:
|
|
task_timeout = TIMEOUT
|
|
start = time.time()
|
|
end = time.time()
|
|
log.info(f"wait bulk load timeout is {task_timeout}")
|
|
pending_tasks = self.get_bulk_insert_pending_list()
|
|
log.info(f"before waiting, there are {len(pending_tasks)} pending tasks")
|
|
while (
|
|
len(tasks_state_distribution["success"]) + len(tasks_state_distribution["failed"]) < len(task_ids)
|
|
and end - start <= task_timeout
|
|
):
|
|
time.sleep(2)
|
|
|
|
for task_id in task_ids:
|
|
if task_id in tasks_state_distribution["success"] or task_id in tasks_state_distribution["failed"]:
|
|
continue
|
|
else:
|
|
state, _ = self.get_bulk_insert_state(task_id, task_timeout, using, **kwargs)
|
|
tasks_state[task_id] = state
|
|
|
|
if target_state == BulkInsertState.ImportPersisted:
|
|
if state.state in [BulkInsertState.ImportPersisted, BulkInsertState.ImportCompleted]:
|
|
if task_id in tasks_state_distribution["in_progress"]:
|
|
tasks_state_distribution["in_progress"].remove(task_id)
|
|
tasks_state_distribution["success"].add(task_id)
|
|
elif state.state in [BulkInsertState.ImportPending, BulkInsertState.ImportStarted]:
|
|
tasks_state_distribution["in_progress"].add(task_id)
|
|
else:
|
|
tasks_state_distribution["failed"].add(task_id)
|
|
|
|
if target_state == BulkInsertState.ImportCompleted:
|
|
if state.state in [BulkInsertState.ImportCompleted]:
|
|
if task_id in tasks_state_distribution["in_progress"]:
|
|
tasks_state_distribution["in_progress"].remove(task_id)
|
|
tasks_state_distribution["success"].add(task_id)
|
|
elif state.state in [
|
|
BulkInsertState.ImportPending,
|
|
BulkInsertState.ImportStarted,
|
|
BulkInsertState.ImportPersisted,
|
|
]:
|
|
tasks_state_distribution["in_progress"].add(task_id)
|
|
else:
|
|
tasks_state_distribution["failed"].add(task_id)
|
|
|
|
end = time.time()
|
|
pending_tasks = self.get_bulk_insert_pending_list()
|
|
log.info(f"after waiting, there are {len(pending_tasks)} pending tasks")
|
|
log.info(f"task state distribution: {tasks_state_distribution}")
|
|
log.info(tasks_state)
|
|
if len(tasks_state_distribution["success"]) == len(task_ids):
|
|
log.info(f"wait for bulk load tasks completed successfully, cost time: {end - start}")
|
|
return True, tasks_state
|
|
else:
|
|
log.info(f"wait for bulk load tasks completed failed, cost time: {end - start}")
|
|
return False, tasks_state
|
|
|
|
def wait_all_pending_tasks_finished(self):
|
|
task_states_map = {}
|
|
all_tasks, _ = self.list_bulk_insert_tasks()
|
|
# log.info(f"all tasks: {all_tasks}")
|
|
for task in all_tasks:
|
|
if task.state in [BulkInsertState.ImportStarted, BulkInsertState.ImportPersisted]:
|
|
task_states_map[task.task_id] = task.state
|
|
|
|
log.info(f"current tasks states: {task_states_map}")
|
|
pending_tasks = self.get_bulk_insert_pending_list()
|
|
working_tasks = self.get_bulk_insert_working_list()
|
|
log.info(
|
|
f"in the start, there are {len(working_tasks)} working tasks, {working_tasks} {len(pending_tasks)} pending tasks, {pending_tasks}"
|
|
)
|
|
time_cnt = 0
|
|
pending_task_ids = set()
|
|
while len(pending_tasks) > 0:
|
|
time.sleep(5)
|
|
time_cnt += 5
|
|
pending_tasks = self.get_bulk_insert_pending_list()
|
|
working_tasks = self.get_bulk_insert_working_list()
|
|
cur_pending_task_ids = []
|
|
for task_id in pending_tasks.keys():
|
|
cur_pending_task_ids.append(task_id)
|
|
pending_task_ids.add(task_id)
|
|
log.info(
|
|
f"after {time_cnt}, there are {len(working_tasks)} working tasks, {len(pending_tasks)} pending tasks"
|
|
)
|
|
log.debug(f"total pending tasks: {pending_task_ids} current pending tasks: {cur_pending_task_ids}")
|
|
log.info(f"after {time_cnt}, all pending tasks are finished")
|
|
all_tasks, _ = self.list_bulk_insert_tasks()
|
|
for task in all_tasks:
|
|
if task.task_id in pending_task_ids:
|
|
log.info(f"task {task.task_id} state transfer from pending to {task.state_name}")
|
|
|
|
def wait_index_build_completed(self, collection_name, timeout=None):
|
|
start = time.time()
|
|
if timeout is not None:
|
|
task_timeout = timeout
|
|
else:
|
|
task_timeout = TIMEOUT
|
|
end = time.time()
|
|
while end - start <= task_timeout:
|
|
time.sleep(0.5)
|
|
index_states, _ = self.index_building_progress(collection_name)
|
|
log.debug(f"index states: {index_states}")
|
|
if index_states["total_rows"] == index_states["indexed_rows"]:
|
|
log.info("index build completed")
|
|
return True
|
|
end = time.time()
|
|
log.info("index build timeout")
|
|
return False
|
|
|
|
def get_query_segment_info(self, collection_name, timeout=None, using="default", check_task=None, check_items=None):
|
|
timeout = TIMEOUT if timeout is None else timeout
|
|
func_name = sys._getframe().f_code.co_name
|
|
res, is_succ = api_request([self.ut.get_query_segment_info, collection_name, timeout, using])
|
|
check_result = ResponseChecker(
|
|
res,
|
|
func_name,
|
|
check_task,
|
|
check_items,
|
|
is_succ,
|
|
collection_name=collection_name,
|
|
timeout=timeout,
|
|
using=using,
|
|
).run()
|
|
return res, check_result
|
|
|
|
def loading_progress(
|
|
self, collection_name, partition_names=None, using="default", check_task=None, check_items=None
|
|
):
|
|
func_name = sys._getframe().f_code.co_name
|
|
res, is_succ = api_request([self.ut.loading_progress, collection_name, partition_names, using])
|
|
check_result = ResponseChecker(
|
|
res,
|
|
func_name,
|
|
check_task,
|
|
check_items,
|
|
is_succ,
|
|
collection_name=collection_name,
|
|
partition_names=partition_names,
|
|
using=using,
|
|
).run()
|
|
return res, check_result
|
|
|
|
def load_state(self, collection_name, partition_names=None, using="default", check_task=None, check_items=None):
|
|
func_name = sys._getframe().f_code.co_name
|
|
res, is_succ = api_request([self.ut.load_state, collection_name, partition_names, using])
|
|
check_result = ResponseChecker(
|
|
res,
|
|
func_name,
|
|
check_task,
|
|
check_items,
|
|
is_succ,
|
|
collection_name=collection_name,
|
|
partition_names=partition_names,
|
|
using=using,
|
|
).run()
|
|
return res, check_result
|
|
|
|
def wait_for_loading_complete(
|
|
self, collection_name, partition_names=None, timeout=None, using="default", check_task=None, check_items=None
|
|
):
|
|
timeout = TIMEOUT if timeout is None else timeout
|
|
|
|
func_name = sys._getframe().f_code.co_name
|
|
res, is_succ = api_request(
|
|
[self.ut.wait_for_loading_complete, collection_name, partition_names, timeout, using]
|
|
)
|
|
check_result = ResponseChecker(
|
|
res,
|
|
func_name,
|
|
check_task,
|
|
check_items,
|
|
is_succ,
|
|
collection_name=collection_name,
|
|
partition_names=partition_names,
|
|
timeout=timeout,
|
|
using=using,
|
|
).run()
|
|
return res, check_result
|
|
|
|
def index_building_progress(
|
|
self, collection_name, index_name="", using="default", check_task=None, check_items=None
|
|
):
|
|
func_name = sys._getframe().f_code.co_name
|
|
res, is_succ = api_request([self.ut.index_building_progress, collection_name, index_name, using])
|
|
check_result = ResponseChecker(
|
|
res,
|
|
func_name,
|
|
check_task,
|
|
check_items,
|
|
is_succ,
|
|
collection_name=collection_name,
|
|
index_name=index_name,
|
|
using=using,
|
|
).run()
|
|
return res, check_result
|
|
|
|
def wait_for_index_building_complete(
|
|
self, collection_name, index_name="", timeout=None, using="default", check_task=None, check_items=None
|
|
):
|
|
timeout = TIMEOUT if timeout is None else timeout
|
|
|
|
func_name = sys._getframe().f_code.co_name
|
|
res, is_succ = api_request(
|
|
[self.ut.wait_for_index_building_complete, collection_name, index_name, timeout, using]
|
|
)
|
|
check_result = ResponseChecker(
|
|
res,
|
|
func_name,
|
|
check_task,
|
|
check_items,
|
|
is_succ,
|
|
collection_name=collection_name,
|
|
index_name=index_name,
|
|
timeout=timeout,
|
|
using=using,
|
|
).run()
|
|
return res, check_result
|
|
|
|
def has_collection(self, collection_name, using="default", check_task=None, check_items=None):
|
|
func_name = sys._getframe().f_code.co_name
|
|
res, is_succ = api_request([self.ut.has_collection, collection_name, using])
|
|
check_result = ResponseChecker(
|
|
res, func_name, check_task, check_items, is_succ, collection_name=collection_name, using=using
|
|
).run()
|
|
return res, check_result
|
|
|
|
def has_partition(self, collection_name, partition_name, using="default", check_task=None, check_items=None):
|
|
func_name = sys._getframe().f_code.co_name
|
|
res, is_succ = api_request([self.ut.has_partition, collection_name, partition_name, using])
|
|
check_result = ResponseChecker(
|
|
res,
|
|
func_name,
|
|
check_task,
|
|
check_items,
|
|
is_succ,
|
|
collection_name=collection_name,
|
|
partition_name=partition_name,
|
|
using=using,
|
|
).run()
|
|
return res, check_result
|
|
|
|
def drop_collection(self, collection_name, timeout=None, using="default", check_task=None, check_items=None):
|
|
func_name = sys._getframe().f_code.co_name
|
|
res, is_succ = api_request([self.ut.drop_collection, collection_name, timeout, using])
|
|
check_result = ResponseChecker(
|
|
res,
|
|
func_name,
|
|
check_task,
|
|
check_items,
|
|
is_succ,
|
|
collection_name=collection_name,
|
|
timeout=timeout,
|
|
using=using,
|
|
).run()
|
|
return res, check_result
|
|
|
|
def list_collections(self, timeout=None, using="default", check_task=None, check_items=None):
|
|
timeout = TIMEOUT if timeout is None else timeout
|
|
|
|
func_name = sys._getframe().f_code.co_name
|
|
res, is_succ = api_request([self.ut.list_collections, timeout, using])
|
|
check_result = ResponseChecker(
|
|
res, func_name, check_task, check_items, is_succ, timeout=timeout, using=using
|
|
).run()
|
|
return res, check_result
|
|
|
|
def calc_distance(
|
|
self, vectors_left, vectors_right, params=None, timeout=None, using="default", check_task=None, check_items=None
|
|
):
|
|
timeout = TIMEOUT if timeout is None else timeout
|
|
|
|
func_name = sys._getframe().f_code.co_name
|
|
res, is_succ = api_request([self.ut.calc_distance, vectors_left, vectors_right, params, timeout, using])
|
|
check_result = ResponseChecker(
|
|
res, func_name, check_task, check_items, is_succ, timeout=timeout, using=using
|
|
).run()
|
|
return res, check_result
|
|
|
|
def load_balance(
|
|
self,
|
|
collection_name,
|
|
src_node_id,
|
|
dst_node_ids,
|
|
sealed_segment_ids,
|
|
timeout=None,
|
|
using="default",
|
|
check_task=None,
|
|
check_items=None,
|
|
):
|
|
timeout = TIMEOUT if timeout is None else timeout
|
|
|
|
func_name = sys._getframe().f_code.co_name
|
|
res, is_succ = api_request(
|
|
[self.ut.load_balance, collection_name, src_node_id, dst_node_ids, sealed_segment_ids, timeout, using]
|
|
)
|
|
check_result = ResponseChecker(
|
|
res, func_name, check_task, check_items, is_succ, timeout=timeout, using=using
|
|
).run()
|
|
return res, check_result
|
|
|
|
def create_alias(self, collection_name, alias, timeout=None, using="default", check_task=None, check_items=None):
|
|
timeout = TIMEOUT if timeout is None else timeout
|
|
func_name = sys._getframe().f_code.co_name
|
|
res, is_succ = api_request([self.ut.create_alias, collection_name, alias, timeout, using])
|
|
check_result = ResponseChecker(
|
|
res, func_name, check_task, check_items, is_succ, timeout=timeout, using=using
|
|
).run()
|
|
return res, check_result
|
|
|
|
def drop_alias(self, alias, timeout=None, using="default", check_task=None, check_items=None):
|
|
timeout = TIMEOUT if timeout is None else timeout
|
|
func_name = sys._getframe().f_code.co_name
|
|
res, is_succ = api_request([self.ut.drop_alias, alias, timeout, using])
|
|
check_result = ResponseChecker(
|
|
res, func_name, check_task, check_items, is_succ, timeout=timeout, using=using
|
|
).run()
|
|
return res, check_result
|
|
|
|
def alter_alias(self, collection_name, alias, timeout=None, using="default", check_task=None, check_items=None):
|
|
timeout = TIMEOUT if timeout is None else timeout
|
|
func_name = sys._getframe().f_code.co_name
|
|
res, is_succ = api_request([self.ut.alter_alias, collection_name, alias, timeout, using])
|
|
check_result = ResponseChecker(
|
|
res, func_name, check_task, check_items, is_succ, timeout=timeout, using=using
|
|
).run()
|
|
return res, check_result
|
|
|
|
def list_aliases(self, collection_name, timeout=None, using="default", check_task=None, check_items=None):
|
|
timeout = TIMEOUT if timeout is None else timeout
|
|
func_name = sys._getframe().f_code.co_name
|
|
res, is_succ = api_request([self.ut.list_aliases, collection_name, timeout, using])
|
|
check_result = ResponseChecker(
|
|
res, func_name, check_task, check_items, is_succ, timeout=timeout, using=using
|
|
).run()
|
|
return res, check_result
|
|
|
|
def mkts_from_datetime(self, d_time=None, milliseconds=0.0, delta=None):
|
|
d_time = datetime.now() if d_time is None else d_time
|
|
res, _ = api_request([self.ut.mkts_from_datetime, d_time, milliseconds, delta])
|
|
return res
|
|
|
|
def mkts_from_hybridts(self, hybridts, milliseconds=0.0, delta=None):
|
|
res, _ = api_request([self.ut.mkts_from_hybridts, hybridts, milliseconds, delta])
|
|
return res
|
|
|
|
def create_user(self, user, password, using="default", check_task=None, check_items=None):
|
|
func_name = sys._getframe().f_code.co_name
|
|
res, is_succ = api_request([self.ut.create_user, user, password, using])
|
|
check_result = ResponseChecker(res, func_name, check_task, check_items, is_succ, using=using).run()
|
|
return res, check_result
|
|
|
|
def list_usernames(self, using="default", check_task=None, check_items=None):
|
|
func_name = sys._getframe().f_code.co_name
|
|
res, is_succ = api_request([self.ut.list_usernames, using])
|
|
check_result = ResponseChecker(res, func_name, check_task, check_items, is_succ, using=using).run()
|
|
return res, check_result
|
|
|
|
def reset_password(self, user, old_password, new_password, check_task=None, check_items=None):
|
|
func_name = sys._getframe().f_code.co_name
|
|
res, is_succ = api_request([self.ut.reset_password, user, old_password, new_password])
|
|
check_result = ResponseChecker(res, func_name, check_task, check_items, is_succ).run()
|
|
return res, check_result
|
|
|
|
def update_password(self, user, old_password, new_password, check_task=None, check_items=None):
|
|
func_name = sys._getframe().f_code.co_name
|
|
res, is_succ = api_request([self.ut.update_password, user, old_password, new_password])
|
|
check_result = ResponseChecker(res, func_name, check_task, check_items, is_succ).run()
|
|
return res, check_result
|
|
|
|
def delete_user(self, user, using="default", check_task=None, check_items=None):
|
|
func_name = sys._getframe().f_code.co_name
|
|
res, is_succ = api_request([self.ut.delete_user, user, using])
|
|
check_result = ResponseChecker(res, func_name, check_task, check_items, is_succ, using=using).run()
|
|
return res, check_result
|
|
|
|
def list_roles(self, include_user_info: bool, using="default", check_task=None, check_items=None):
|
|
func_name = sys._getframe().f_code.co_name
|
|
res, is_succ = api_request([self.ut.list_roles, include_user_info, using])
|
|
check_result = ResponseChecker(res, func_name, check_task, check_items, is_succ, using=using).run()
|
|
return res, check_result
|
|
|
|
def list_user(self, username: str, include_role_info: bool, using="default", check_task=None, check_items=None):
|
|
func_name = sys._getframe().f_code.co_name
|
|
res, is_succ = api_request([self.ut.list_user, username, include_role_info, using])
|
|
check_result = ResponseChecker(res, func_name, check_task, check_items, is_succ, using=using).run()
|
|
return res, check_result
|
|
|
|
def list_users(self, include_role_info: bool, using="default", check_task=None, check_items=None):
|
|
func_name = sys._getframe().f_code.co_name
|
|
res, is_succ = api_request([self.ut.list_users, include_role_info, using])
|
|
check_result = ResponseChecker(res, func_name, check_task, check_items, is_succ, using=using).run()
|
|
return res, check_result
|
|
|
|
def init_role(self, name, using="default", check_task=None, check_items=None, **kwargs):
|
|
func_name = sys._getframe().f_code.co_name
|
|
res, is_succ = api_request([Role, name, using], **kwargs)
|
|
self.role = res if is_succ else None
|
|
check_result = ResponseChecker(res, func_name, check_task, check_items, is_succ, name=name, **kwargs).run()
|
|
return res, check_result
|
|
|
|
def create_role(self, check_task=None, check_items=None, **kwargs):
|
|
func_name = sys._getframe().f_code.co_name
|
|
res, is_succ = api_request([self.role.create], **kwargs)
|
|
check_result = ResponseChecker(res, func_name, check_task, check_items, is_succ, **kwargs).run()
|
|
return res, check_result
|
|
|
|
def role_drop(self, check_task=None, check_items=None, **kwargs):
|
|
func_name = sys._getframe().f_code.co_name
|
|
res, check = api_request([self.role.drop], **kwargs)
|
|
check_result = ResponseChecker(res, func_name, check_task, check_items, check, **kwargs).run()
|
|
return res, check_result
|
|
|
|
def role_is_exist(self, check_task=None, check_items=None, **kwargs):
|
|
func_name = sys._getframe().f_code.co_name
|
|
res, check = api_request([self.role.is_exist], **kwargs)
|
|
check_result = ResponseChecker(res, func_name, check_task, check_items, check, **kwargs).run()
|
|
return res, check_result
|
|
|
|
def role_add_user(self, username: str, check_task=None, check_items=None, **kwargs):
|
|
func_name = sys._getframe().f_code.co_name
|
|
res, check = api_request([self.role.add_user, username], **kwargs)
|
|
check_result = ResponseChecker(res, func_name, check_task, check_items, check, **kwargs).run()
|
|
return res, check_result
|
|
|
|
def role_remove_user(self, username: str, check_task=None, check_items=None, **kwargs):
|
|
func_name = sys._getframe().f_code.co_name
|
|
res, check = api_request([self.role.remove_user, username], **kwargs)
|
|
check_result = ResponseChecker(res, func_name, check_task, check_items, check, **kwargs).run()
|
|
return res, check_result
|
|
|
|
def role_get_users(self, check_task=None, check_items=None, **kwargs):
|
|
func_name = sys._getframe().f_code.co_name
|
|
res, check = api_request([self.role.get_users], **kwargs)
|
|
check_result = ResponseChecker(res, func_name, check_task, check_items, check, **kwargs).run()
|
|
return res, check_result
|
|
|
|
@property
|
|
def role_name(self):
|
|
return self.role.name
|
|
|
|
def role_grant(
|
|
self,
|
|
object: str,
|
|
object_name: str,
|
|
privilege: str,
|
|
db_name: str = "",
|
|
check_task=None,
|
|
check_items=None,
|
|
**kwargs,
|
|
):
|
|
func_name = sys._getframe().f_code.co_name
|
|
res, check = api_request([self.role.grant, object, object_name, privilege, db_name], **kwargs)
|
|
check_result = ResponseChecker(res, func_name, check_task, check_items, check, **kwargs).run()
|
|
return res, check_result
|
|
|
|
def role_revoke(
|
|
self,
|
|
object: str,
|
|
object_name: str,
|
|
privilege: str,
|
|
db_name: str = "",
|
|
check_task=None,
|
|
check_items=None,
|
|
**kwargs,
|
|
):
|
|
func_name = sys._getframe().f_code.co_name
|
|
res, check = api_request([self.role.revoke, object, object_name, privilege, db_name], **kwargs)
|
|
check_result = ResponseChecker(res, func_name, check_task, check_items, check, **kwargs).run()
|
|
return res, check_result
|
|
|
|
def role_grant_v2(
|
|
self, privilege: str, collection_name: str, db_name: str = None, check_task=None, check_items=None, **kwargs
|
|
):
|
|
func_name = sys._getframe().f_code.co_name
|
|
res, check = api_request([self.role.grant_v2, privilege, collection_name, db_name], **kwargs)
|
|
check_result = ResponseChecker(res, func_name, check_task, check_items, check, **kwargs).run()
|
|
return res, check_result
|
|
|
|
def role_revoke_v2(
|
|
self, privilege: str, collection_name: str, db_name: str = None, check_task=None, check_items=None, **kwargs
|
|
):
|
|
func_name = sys._getframe().f_code.co_name
|
|
res, check = api_request([self.role.revoke_v2, privilege, collection_name, db_name], **kwargs)
|
|
check_result = ResponseChecker(res, func_name, check_task, check_items, check, **kwargs).run()
|
|
return res, check_result
|
|
|
|
def role_list_grant(
|
|
self, object: str, object_name: str, db_name: str = "", check_task=None, check_items=None, **kwargs
|
|
):
|
|
func_name = sys._getframe().f_code.co_name
|
|
res, check = api_request([self.role.list_grant, object, object_name, db_name], **kwargs)
|
|
check_result = ResponseChecker(res, func_name, check_task, check_items, check, **kwargs).run()
|
|
return res, check_result
|
|
|
|
def role_list_grants(self, db_name: str = "", check_task=None, check_items=None, **kwargs):
|
|
func_name = sys._getframe().f_code.co_name
|
|
res, check = api_request([self.role.list_grants, db_name], **kwargs)
|
|
check_result = ResponseChecker(res, func_name, check_task, check_items, check, **kwargs).run()
|
|
return res, check_result
|
|
|
|
def create_resource_group(self, name, using="default", timeout=None, check_task=None, check_items=None, **kwargs):
|
|
func_name = sys._getframe().f_code.co_name
|
|
res, check = api_request([self.ut.create_resource_group, name, using, timeout], **kwargs)
|
|
check_result = ResponseChecker(res, func_name, check_task, check_items, check, **kwargs).run()
|
|
return res, check_result
|
|
|
|
def drop_resource_group(self, name, using="default", timeout=None, check_task=None, check_items=None, **kwargs):
|
|
func_name = sys._getframe().f_code.co_name
|
|
res, check = api_request([self.ut.drop_resource_group, name, using, timeout], **kwargs)
|
|
check_result = ResponseChecker(res, func_name, check_task, check_items, check, **kwargs).run()
|
|
return res, check_result
|
|
|
|
def list_resource_groups(self, using="default", timeout=None, check_task=None, check_items=None, **kwargs):
|
|
func_name = sys._getframe().f_code.co_name
|
|
res, check = api_request([self.ut.list_resource_groups, using, timeout], **kwargs)
|
|
check_result = ResponseChecker(res, func_name, check_task, check_items, check, **kwargs).run()
|
|
return res, check_result
|
|
|
|
def describe_resource_group(self, name, using="default", timeout=None, check_task=None, check_items=None, **kwargs):
|
|
func_name = sys._getframe().f_code.co_name
|
|
res, check = api_request([self.ut.describe_resource_group, name, using, timeout], **kwargs)
|
|
check_result = ResponseChecker(res, func_name, check_task, check_items, check, **kwargs).run()
|
|
return res, check_result
|
|
|
|
def update_resource_group(self, config, using="default", timeout=None, check_task=None, check_items=None, **kwargs):
|
|
func_name = sys._getframe().f_code.co_name
|
|
res, check = api_request([self.ut.update_resource_groups, config, using, timeout], **kwargs)
|
|
check_result = ResponseChecker(res, func_name, check_task, check_items, check, **kwargs).run()
|
|
return res, check_result
|
|
|
|
def transfer_node(
|
|
self, source, target, num_node, using="default", timeout=None, check_task=None, check_items=None, **kwargs
|
|
):
|
|
func_name = sys._getframe().f_code.co_name
|
|
res, check = api_request([self.ut.transfer_node, source, target, num_node, using, timeout], **kwargs)
|
|
check_result = ResponseChecker(res, func_name, check_task, check_items, check, **kwargs).run()
|
|
return res, check_result
|
|
|
|
def transfer_replica(
|
|
self,
|
|
source,
|
|
target,
|
|
collection_name,
|
|
num_replica,
|
|
using="default",
|
|
timeout=None,
|
|
check_task=None,
|
|
check_items=None,
|
|
**kwargs,
|
|
):
|
|
func_name = sys._getframe().f_code.co_name
|
|
res, check = api_request(
|
|
[self.ut.transfer_replica, source, target, collection_name, num_replica, using, timeout], **kwargs
|
|
)
|
|
check_result = ResponseChecker(res, func_name, check_task, check_items, check, **kwargs).run()
|
|
return res, check_result
|
|
|
|
def rename_collection(
|
|
self,
|
|
old_collection_name,
|
|
new_collection_name,
|
|
new_db_name="",
|
|
timeout=None,
|
|
check_task=None,
|
|
check_items=None,
|
|
**kwargs,
|
|
):
|
|
func_name = sys._getframe().f_code.co_name
|
|
res, check = api_request(
|
|
[self.ut.rename_collection, old_collection_name, new_collection_name, new_db_name, timeout], **kwargs
|
|
)
|
|
check_result = ResponseChecker(
|
|
res,
|
|
func_name,
|
|
check_task,
|
|
check_items,
|
|
check,
|
|
old_collection_name=old_collection_name,
|
|
new_collection_name=new_collection_name,
|
|
new_db_name=new_db_name,
|
|
timeout=timeout,
|
|
**kwargs,
|
|
).run()
|
|
return res, check_result
|
|
|
|
def flush_all(self, using="default", timeout=None, check_task=None, check_items=None, **kwargs):
|
|
func_name = sys._getframe().f_code.co_name
|
|
res, check = api_request([self.ut.flush_all, using, timeout], **kwargs)
|
|
check_result = ResponseChecker(
|
|
res, func_name, check_task, check_items, check, using=using, timeout=timeout, **kwargs
|
|
).run()
|
|
return res, check_result
|
|
|
|
def get_server_type(self, using="default", check_task=None, check_items=None, **kwargs):
|
|
func_name = sys._getframe().f_code.co_name
|
|
res, check = api_request([self.ut.get_server_type, using], **kwargs)
|
|
check_result = ResponseChecker(res, func_name, check_task, check_items, check, using=using, **kwargs).run()
|
|
return res, check_result
|
|
|
|
def list_indexes(self, collection_name, using="default", timeout=None, check_task=None, check_items=None, **kwargs):
|
|
func_name = sys._getframe().f_code.co_name
|
|
res, check = api_request([self.ut.list_indexes, collection_name, using, timeout], **kwargs)
|
|
check_result = ResponseChecker(
|
|
res,
|
|
func_name,
|
|
check_task,
|
|
check_items,
|
|
check,
|
|
collection_name=collection_name,
|
|
using=using,
|
|
timeout=timeout,
|
|
**kwargs,
|
|
).run()
|
|
return res, check_result
|
|
|
|
def create_privilege_group(self, privilege_group: str, check_task=None, check_items=None, **kwargs):
|
|
func_name = sys._getframe().f_code.co_name
|
|
res, check = api_request([self.role.create_privilege_group, privilege_group], **kwargs)
|
|
check_result = ResponseChecker(res, func_name, check_task, check_items, check, **kwargs).run()
|
|
return res, check_result
|
|
|
|
def drop_privilege_group(self, privilege_group: str, check_task=None, check_items=None, **kwargs):
|
|
func_name = sys._getframe().f_code.co_name
|
|
res, check = api_request([self.role.drop_privilege_group, privilege_group], **kwargs)
|
|
check_result = ResponseChecker(res, func_name, check_task, check_items, check, **kwargs).run()
|
|
return res, check_result
|
|
|
|
def list_privilege_groups(self, check_task=None, check_items=None, **kwargs):
|
|
func_name = sys._getframe().f_code.co_name
|
|
res, check = api_request([self.role.list_privilege_groups], **kwargs)
|
|
check_result = ResponseChecker(res, func_name, check_task, check_items, check, **kwargs).run()
|
|
return res, check_result
|
|
|
|
def add_privileges_to_group(
|
|
self, privilege_group: str, privileges: list, check_task=None, check_items=None, **kwargs
|
|
):
|
|
func_name = sys._getframe().f_code.co_name
|
|
res, check = api_request([self.role.add_privileges_to_group, privilege_group, privileges], **kwargs)
|
|
check_result = ResponseChecker(res, func_name, check_task, check_items, check, **kwargs).run()
|
|
return res, check_result
|
|
|
|
def remove_privileges_from_group(
|
|
self, privilege_group: str, privileges: list, check_task=None, check_items=None, **kwargs
|
|
):
|
|
func_name = sys._getframe().f_code.co_name
|
|
res, check = api_request([self.role.remove_privileges_from_group, privilege_group, privileges], **kwargs)
|
|
check_result = ResponseChecker(res, func_name, check_task, check_items, check, **kwargs).run()
|
|
return res, check_result
|