mirrors/milvus
0
0 Fork 0
mirror of https://github.com/milvus-io/milvus.git synced 2026-10-29 09:25:54 +08:00
Code Activity
Files
master
milvus/tests/python_client/testcases/test_alias.py
T
JamesxiaofanluanClaude Opus 5
 f929d68ba6 test: clear the ruff baseline in tests/ so the lint gate reflects the PR under review (#52115) 
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>
2026-08-03 14:54:18 +08:00

490 lines
25 KiB
Python
Raw Permalink Blame History

##################################################################
# All test cases in this file have been migrated to milvus_client#
##################################################################
'''
import pytest
import random
from base.client_base import TestcaseBase
from utils.util_log import test_log as log
from common import common_func as cf
from common import common_type as ct
from common.common_type import CaseLabel, CheckTasks
prefix = "alias"
exp_name = "name"
exp_schema = "schema"
default_schema = cf.gen_default_collection_schema()
default_binary_schema = cf.gen_default_binary_collection_schema()
default_nb = ct.default_nb
default_nb_medium = ct.default_nb_medium
default_nq = ct.default_nq
default_dim = ct.default_dim
default_limit = ct.default_limit
default_search_exp = "int64 >= 0"
default_search_field = ct.default_float_vec_field_name
default_search_params = ct.default_search_params
class TestAliasParamsInvalid(TestcaseBase):
""" Negative test cases of alias interface parameters"""
@pytest.mark.tags(CaseLabel.L2)
@pytest.mark.parametrize("alias_name", ["12-s", "12 s", "(mn)", "中文", "%$#", "a".join("a" for i in range(256))])
def test_alias_create_alias_with_invalid_name(self, alias_name):
"""
target: test alias inserting data
method: create a collection with invalid alias name
expected: create alias failed
"""
self._connect()
c_name = cf.gen_unique_str("collection")
collection_w = self.init_collection_wrap(name=c_name, schema=default_schema,
check_task=CheckTasks.check_collection_property,
check_items={exp_name: c_name, exp_schema: default_schema})
error = {ct.err_code: 1100, ct.err_msg: "Invalid collection alias"}
self.utility_wrap.create_alias(collection_w.name, alias_name,
check_task=CheckTasks.err_res,
check_items=error)
class TestAliasOperation(TestcaseBase):
""" Test cases of alias interface operations"""
@pytest.mark.tags(CaseLabel.L0)
def test_alias_alter_operation_default(self):
"""
target: test collection altering alias
method:
1. create collection_1, bind alias to collection_1 and insert 2000 entities
2. create collection_2 with 1500 entities
3. search on alias
verify num_entities=2000
4. alter alias to collection_2 and search on alias
verify num_entities=1500
"""
c_name1 = cf.gen_unique_str("collection1")
collection_w1 = self.init_collection_wrap(name=c_name1, schema=default_schema,
check_task=CheckTasks.check_collection_property,
check_items={exp_name: c_name1, exp_schema: default_schema})
alias_name = cf.gen_unique_str(prefix)
# create a collection alias and bind to collection1
self.utility_wrap.create_alias(collection_w1.name, alias_name)
collection_alias = self.init_collection_wrap(name=alias_name)
nb1 = 2000
data1 = cf.gen_default_dataframe_data(nb=nb1)
import pandas as pd
string_values = pd.Series(data=[str(i) for i in range(nb1)], dtype="string")
data1[ct.default_string_field_name] = string_values
collection_alias.insert(data1)
collection_alias.create_index(ct.default_float_vec_field_name, ct.default_index)
collection_alias.load()
assert collection_alias.num_entities == nb1 == collection_w1.num_entities
res1 = collection_alias.query(expr="", output_fields=["count(*)"])[0]
assert res1[0].get("count(*)") == nb1
# create collection2
c_name2 = cf.gen_unique_str("collection2")
collection_w2 = self.init_collection_wrap(name=c_name2, schema=default_schema,
check_task=CheckTasks.check_collection_property,
check_items={exp_name: c_name2, exp_schema: default_schema})
nb2 = 1500
data2 = cf.gen_default_dataframe_data(nb=nb2)
string_values = pd.Series(data=[str(i) for i in range(nb2)], dtype="string")
data2[ct.default_string_field_name] = string_values
collection_w2.insert(data2)
collection_w2.create_index(ct.default_float_vec_field_name, ct.default_index)
collection_w2.load()
# alter the collection alias to collection2
self.utility_wrap.alter_alias(collection_w2.name, alias_name)
assert collection_alias.num_entities == nb2 == collection_w2.num_entities
res1 = collection_alias.query(expr="", output_fields=["count(*)"])[0]
assert res1[0].get("count(*)") == nb2
@pytest.mark.tags(CaseLabel.L1)
def test_alias_create_operation_default(self):
"""
target: test collection creating alias
method:
1.create a collection and create 10 partitions for it
2.collection create an alias, then init a collection with this alias but not create partitions
expected: collection is equal to alias
"""
self._connect()
c_name = cf.gen_unique_str("collection")
collection_w = self.init_collection_wrap(name=c_name, schema=default_schema,
check_task=CheckTasks.check_collection_property,
check_items={exp_name: c_name, exp_schema: default_schema})
for _ in range(10):
partition_name = cf.gen_unique_str("partition")
# create partition with different names and check the partition exists
self.init_partition_wrap(collection_w, partition_name)
assert collection_w.has_partition(partition_name)[0]
alias_name = cf.gen_unique_str(prefix)
self.utility_wrap.create_alias(collection_w.name, alias_name)
collection_alias = self.init_collection_wrap(name=alias_name,
check_task=CheckTasks.check_collection_property,
check_items={exp_name: alias_name, exp_schema: default_schema})
# assert collection is equal to alias according to partitions
assert [p.name for p in collection_w.partitions] == [
p.name for p in collection_alias.partitions]
@pytest.mark.tags(CaseLabel.L1)
def test_alias_drop_operation_default(self):
"""
target: test collection dropping alias
method:
1.create a collection with 10 partitions
2.collection create an alias
3.collection drop the alias
expected:
after step 2, collection is equal to alias
after step 3, collection with alias name is not exist
"""
self._connect()
c_name = cf.gen_unique_str("collection")
collection_w = self.init_collection_wrap(name=c_name, schema=default_schema,
check_task=CheckTasks.check_collection_property,
check_items={exp_name: c_name, exp_schema: default_schema})
for _ in range(10):
partition_name = cf.gen_unique_str("partition")
# create partition with different names and check the partition exists
self.init_partition_wrap(collection_w, partition_name)
assert collection_w.has_partition(partition_name)[0]
alias_name = cf.gen_unique_str(prefix)
self.utility_wrap.create_alias(collection_w.name, alias_name)
# collection_w.create_alias(alias_name)
collection_alias = self.init_collection_wrap(name=alias_name,
check_task=CheckTasks.check_collection_property,
check_items={exp_name: alias_name, exp_schema: default_schema})
# assert collection is equal to alias according to partitions
assert [p.name for p in collection_w.partitions] == [
p.name for p in collection_alias.partitions]
self.utility_wrap.drop_alias(alias_name)
error = {ct.err_code: 0,
ct.err_msg: f"Collection '{alias_name}' not exist, or you can pass in schema to create one"}
collection_alias, _ = self.collection_wrap.init_collection(name=alias_name,
check_task=CheckTasks.err_res,
check_items=error)
@pytest.mark.tags(CaseLabel.L2)
def test_alias_called_by_utility_has_collection(self):
"""
target: test utility has collection by alias
method:
1.create collection with alias
2.call has_collection function with alias as param
expected: result is True
"""
self._connect()
c_name = cf.gen_unique_str("collection")
collection_w = self.init_collection_wrap(name=c_name, schema=default_schema,
check_task=CheckTasks.check_collection_property,
check_items={exp_name: c_name, exp_schema: default_schema})
alias_name = cf.gen_unique_str(prefix)
self.utility_wrap.create_alias(collection_w.name, alias_name)
# collection_w.create_alias(alias_name)
collection_alias, _ = self.collection_wrap.init_collection(name=alias_name,
check_task=CheckTasks.check_collection_property,
check_items={exp_name: alias_name,
exp_schema: default_schema})
res, _ = self.utility_wrap.has_collection(alias_name)
assert res is True
@pytest.mark.tags(CaseLabel.L2)
def test_alias_called_by_utility_drop_collection(self):
"""
target: test utility drop collection by alias
method:
1.create collection with alias
2.call drop_collection function with alias as param
expected: Got error: collection cannot be dropped via alias.
"""
self._connect()
c_name = cf.gen_unique_str("collection")
collection_w = self.init_collection_wrap(name=c_name, schema=default_schema,
check_task=CheckTasks.check_collection_property,
check_items={exp_name: c_name, exp_schema: default_schema})
alias_name = cf.gen_unique_str(prefix)
self.utility_wrap.create_alias(collection_w.name, alias_name)
# collection_w.create_alias(alias_name)
collection_alias, _ = self.collection_wrap.init_collection(name=alias_name,
check_task=CheckTasks.check_collection_property,
check_items={exp_name: alias_name,
exp_schema: default_schema})
assert self.utility_wrap.has_collection(c_name)[0]
error = {ct.err_code: 1,
ct.err_msg: f"cannot drop the collection via alias = {alias_name}"}
self.utility_wrap.drop_collection(alias_name,
check_task=CheckTasks.err_res,
check_items=error)
self.utility_wrap.drop_alias(alias_name)
self.utility_wrap.drop_collection(c_name)
assert not self.utility_wrap.has_collection(c_name)[0]
@pytest.mark.tags(CaseLabel.L2)
def test_alias_called_by_utility_has_partition(self):
"""
target: test utility has partition by alias
method:
1.create collection with partition and alias
2.call has_partition function with alias as param
expected: result is True
"""
self._connect()
c_name = cf.gen_unique_str("collection")
collection_w = self.init_collection_wrap(name=c_name, schema=default_schema,
check_task=CheckTasks.check_collection_property,
check_items={exp_name: c_name, exp_schema: default_schema})
partition_name = cf.gen_unique_str("partition")
self.init_partition_wrap(collection_w, partition_name)
alias_name = cf.gen_unique_str(prefix)
self.utility_wrap.create_alias(collection_w.name, alias_name)
# collection_w.create_alias(alias_name)
collection_alias, _ = self.collection_wrap.init_collection(name=alias_name,
check_task=CheckTasks.check_collection_property,
check_items={exp_name: alias_name,
exp_schema: default_schema})
res, _ = self.utility_wrap.has_partition(alias_name, partition_name)
assert res is True
@pytest.mark.tags(CaseLabel.L1)
def test_enable_mmap_by_alias(self):
"""
target: enable or disable mmap by alias
method: enable or disable mmap by alias
expected: successfully enable mmap
"""
self._connect()
c_name = cf.gen_unique_str("collection")
collection_w = self.init_collection_wrap(c_name, schema=default_schema)
alias_name = cf.gen_unique_str(prefix)
self.utility_wrap.create_alias(collection_w.name, alias_name)
collection_alias, _ = self.collection_wrap.init_collection(name=alias_name,
check_task=CheckTasks.check_collection_property,
check_items={exp_name: alias_name,
exp_schema: default_schema})
collection_alias.set_properties({'mmap.enabled': True})
pro = collection_w.describe()[0].get("properties")
assert pro["mmap.enabled"] == 'True'
collection_w.set_properties({'mmap.enabled': False})
pro = collection_alias.describe().get("properties")
assert pro["mmap.enabled"] == 'False'
##########################################################
# class TestAliasOperationInvalid() has been migrated to milvus_client
##########################################################
class TestAliasOperationInvalid(TestcaseBase):
""" Negative test cases of alias interface operations"""
@pytest.mark.tags(CaseLabel.L1)
def test_alias_create_duplication_alias(self):
"""
target: test two collections creating alias with same name
method:
1.create a collection_1 with alias name alias_a
2.create a collection_2 also with alias name alias_a
expected:
in step 2, creating alias with a duplication name is not allowed
"""
self._connect()
c_1_name = cf.gen_unique_str("collection")
collection_1 = self.init_collection_wrap(name=c_1_name, schema=default_schema,
check_task=CheckTasks.check_collection_property,
check_items={exp_name: c_1_name, exp_schema: default_schema})
alias_a_name = cf.gen_unique_str(prefix)
self.utility_wrap.create_alias(collection_1.name, alias_a_name)
c_2_name = cf.gen_unique_str("collection")
collection_2 = self.init_collection_wrap(name=c_2_name, schema=default_schema,
check_task=CheckTasks.check_collection_property,
check_items={exp_name: c_2_name, exp_schema: default_schema})
error = {ct.err_code: 1602,
ct.err_msg: f"{alias_a_name} is alias to another collection: {collection_1.name}: "
f"alias already exist[database=default][alias={alias_a_name}]"}
self.utility_wrap.create_alias(collection_2.name, alias_a_name,
check_task=CheckTasks.err_res,
check_items=error)
@pytest.mark.tags(CaseLabel.L1)
def test_alias_alter_not_exist_alias(self):
"""
target: test collection altering to alias which is not exist
method:
1.create a collection with alias
2.collection alters to an alias name which is not exist
expected:
in step 2, alter alias with a not exist name is not allowed
"""
self._connect()
c_name = cf.gen_unique_str("collection")
collection_w = self.init_collection_wrap(name=c_name, schema=default_schema,
check_task=CheckTasks.check_collection_property,
check_items={exp_name: c_name, exp_schema: default_schema})
alias_name = cf.gen_unique_str(prefix)
self.utility_wrap.create_alias(collection_w.name, alias_name)
alias_not_exist_name = cf.gen_unique_str(prefix)
error = {ct.err_code: 1600,
ct.err_msg: f"alias not found[database=default][alias={alias_not_exist_name}]"}
self.utility_wrap.alter_alias(collection_w.name, alias_not_exist_name,
check_task=CheckTasks.err_res,
check_items=error)
@pytest.mark.tags(CaseLabel.L2)
def test_alias_drop_not_exist_alias(self):
"""
target: test collection dropping alias which is not exist
method:
1.create a collection with alias
2.collection drop alias which is not exist
expected: drop alias succ
"""
self._connect()
c_name = cf.gen_unique_str("collection")
collection_w = self.init_collection_wrap(name=c_name, schema=default_schema,
check_task=CheckTasks.check_collection_property,
check_items={exp_name: c_name, exp_schema: default_schema})
alias_name = cf.gen_unique_str(prefix)
self.utility_wrap.create_alias(collection_w.name, alias_name)
alias_not_exist_name = cf.gen_unique_str(prefix)
self.utility_wrap.drop_alias(alias_not_exist_name)
@pytest.mark.tags(CaseLabel.L2)
def test_alias_drop_same_alias_twice(self):
"""
target: test collection dropping same alias twice
method:
1.create a collection with alias
2.collection drop alias
3.collection drop alias again
expected: drop alias succ
"""
self._connect()
c_name = cf.gen_unique_str("collection")
collection_w = self.init_collection_wrap(name=c_name, schema=default_schema,
check_task=CheckTasks.check_collection_property,
check_items={exp_name: c_name, exp_schema: default_schema})
alias_name = cf.gen_unique_str(prefix)
self.utility_wrap.create_alias(collection_w.name, alias_name)
self.utility_wrap.drop_alias(alias_name)
# @longjiquan: dropping alias should be idempotent.
self.utility_wrap.drop_alias(alias_name)
@pytest.mark.tags(CaseLabel.L2)
def test_alias_create_dup_name_collection(self):
"""
target: test creating a collection with a same name as alias, but a different schema
method:
1.create a collection with alias
2.create a collection with same name as alias, but a different schema
expected: in step 2, create collection failed
"""
self._connect()
c_name = cf.gen_unique_str("collection")
collection_w = self.init_collection_wrap(name=c_name, schema=default_schema,
check_task=CheckTasks.check_collection_property,
check_items={exp_name: c_name, exp_schema: default_schema})
alias_name = cf.gen_unique_str(prefix)
self.utility_wrap.create_alias(collection_w.name, alias_name)
# collection_w.create_alias(alias_name)
error = {ct.err_code: 0,
ct.err_msg: "The collection already exist, but the schema is not the same as the schema passed in"}
self.init_collection_wrap(alias_name, schema=default_binary_schema,
check_task=CheckTasks.err_res,
check_items=error)
@pytest.mark.tags(CaseLabel.L2)
def test_alias_drop_collection_by_alias(self):
"""
target: test dropping a collection by alias
method:
1.create a collection with alias
2.drop a collection by alias
expected: in step 2, drop collection by alias failed by design
"""
self._connect()
c_name = cf.gen_unique_str("collection")
schema = cf.gen_default_collection_schema(description="this is for alias decsription")
collection_w = self.init_collection_wrap(name=c_name, schema=schema,
check_task=CheckTasks.check_collection_property,
check_items={exp_name: c_name, exp_schema: schema})
alias_name = cf.gen_unique_str(prefix)
self.utility_wrap.create_alias(collection_w.name, alias_name)
collection_alias = self.init_collection_wrap(name=alias_name, schema=schema,
check_task=CheckTasks.check_collection_property,
check_items={exp_name: alias_name,
exp_schema: schema})
error = {ct.err_code: 999,
ct.err_msg: f"cannot drop the collection via alias = {alias_name}"}
collection_alias.drop(check_task=CheckTasks.err_res, check_items=error)
@pytest.mark.tags(CaseLabel.L0)
def test_alias_reuse_alias_name_from_dropped_collection(self):
"""
target: test dropping a collection which has a alias
method:
1.create a collection
2.create an alias for the collection
3.drop the collection
4.create a new collection
5.create an alias with the same alias name for the new collection
expected: in step 5, create alias with the same name for the new collection succ
"""
self._connect()
c_name = cf.gen_unique_str("collection")
collection_w = self.init_collection_wrap(name=c_name, schema=default_schema,
check_task=CheckTasks.check_collection_property,
check_items={exp_name: c_name, exp_schema: default_schema})
alias_name = cf.gen_unique_str(prefix)
self.utility_wrap.create_alias(collection_w.name, alias_name)
res = self.utility_wrap.list_aliases(c_name)[0]
assert len(res) == 1
# dropping collection that has an alias shall drop the alias as well
self.utility_wrap.drop_alias(alias_name)
collection_w.drop()
collection_w = self.init_collection_wrap(name=c_name, schema=default_schema,
check_task=CheckTasks.check_collection_property,
check_items={exp_name: c_name, exp_schema: default_schema})
res2 = self.utility_wrap.list_aliases(c_name)[0]
assert len(res2) == 0
# the same alias name can be reused for another collection
self.utility_wrap.create_alias(c_name, alias_name)
res2 = self.utility_wrap.list_aliases(c_name)[0]
assert len(res2) == 1
@pytest.mark.tags(CaseLabel.L0)
def test_alias_rename_collection_to_alias_name(self):
"""
target: test renaming a collection to a alias name
method:
1.create a collection
2.create an alias for the collection
3.rename the collection to the alias name
expected: in step 3, rename collection to alias name failed
"""
self._connect()
c_name = cf.gen_unique_str("collection")
collection_w = self.init_collection_wrap(name=c_name, schema=default_schema,
check_task=CheckTasks.check_collection_property,
check_items={exp_name: c_name, exp_schema: default_schema})
alias_name = cf.gen_unique_str(prefix)
self.utility_wrap.create_alias(collection_w.name, alias_name)
error = {ct.err_code: 999,
ct.err_msg: f"cannot rename collection to an existing alias: {alias_name}"}
self.utility_wrap.rename_collection(collection_w.name, alias_name,
check_task=CheckTasks.err_res, check_items=error)
'''
View Git Blame Copy Permalink