test: add REST import 2PC coverage (#50462)

## What type of PR is this?

/kind test

## What this PR does / why we need it:

Adds RESTful import 2PC coverage for Milvus import lifecycle and
CDC-focused scenarios.

This PR adds:
- REST import job helpers for `/describe`, `/commit`, `/abort`, and
state polling
- pytest options for secondary CDC endpoint/object storage/topology
parameters
- a new REST test suite for import 2PC lifecycle, visibility,
commit/abort, delete semantics, TTL, compaction, formats, data types,
indexes, partitions, DML interleaving, CDC, and fault-tolerance gates
- xfail regression coverage for known issues:
  - #50458 REST import commit/abort authorization gap
  - #50459 NaN/Inf FloatVector import/query behavior
  - #50460 REST `options.auto_commit=null` validation gap

## Which issue(s) this PR fixes:

N/A

## Special notes for your reviewer:

The CDC cases require passing secondary cluster REST/MinIO options and
topology arguments. Without those options, CDC-specific cases skip with
an explicit environment precondition message.

The manual compaction-after-commit case is marked non-strict xfail
because compaction may remain `Executing` on current import 2PC test
builds.

## Test result:

```text
python -m py_compile tests/restful_client_v2/conftest.py tests/restful_client_v2/api/milvus.py tests/restful_client_v2/testcases/test_import_2pc_operation.py
ruff check tests/restful_client_v2/api/milvus.py tests/restful_client_v2/conftest.py tests/restful_client_v2/testcases/test_import_2pc_operation.py
python -m pytest --collect-only -q testcases/test_import_2pc_operation.py
141 tests collected in 2.14s
```

---------

Signed-off-by: zhuwenxing <wenxing.zhu@zilliz.com>
This commit is contained in:
zhuwenxing
2026-06-18 19:00:24 +08:00
committed by GitHub
parent 5d6ab96648
commit a95a54e875
33 changed files with 16950 additions and 3147 deletions
+91
View File
@@ -90,6 +90,57 @@ def pytest_addoption(parser):
default="chaos-testing",
help="Kubernetes namespace for Milvus deployment",
)
parser.addoption(
"--import-2pc-workload",
action="store",
default="true",
help="Whether CDC stability suites should include Import 2PC workload",
)
parser.addoption(
"--import-2pc-minio-host",
action="store",
default="",
help="Upstream MinIO host or host:port used by Import 2PC workload",
)
parser.addoption(
"--import-2pc-minio-bucket",
action="store",
default="",
help="Upstream MinIO bucket used by Import 2PC workload",
)
parser.addoption(
"--import-2pc-downstream-minio-host",
action="store",
default="",
help="Downstream MinIO host or host:port used by CDC Import 2PC workload",
)
parser.addoption(
"--import-2pc-downstream-minio-bucket",
action="store",
default="",
help="Downstream MinIO bucket used by CDC Import 2PC workload",
)
parser.addoption(
"--import-2pc-rows",
action="store",
default="20",
help="Rows per Import 2PC checker operation",
)
def _as_bool(value):
if isinstance(value, bool):
return value
return str(value).lower() in ("1", "true", "yes", "y")
def _minio_endpoint(host):
host = (host or "").strip()
if not host:
return ""
if ":" in host:
return host
return f"{host}:9000"
@pytest.fixture(scope="session")
@@ -180,6 +231,46 @@ def milvus_ns(request):
return request.config.getoption("--milvus-ns")
@pytest.fixture(scope="session")
def import_2pc_workload(request):
return _as_bool(request.config.getoption("--import-2pc-workload"))
@pytest.fixture(scope="session")
def import_2pc_minio_endpoint(request):
host = request.config.getoption("--import-2pc-minio-host") or request.config.getoption("--minio_host")
return _minio_endpoint(host)
@pytest.fixture(scope="session")
def import_2pc_minio_bucket(request):
return request.config.getoption("--import-2pc-minio-bucket") or request.config.getoption("--minio_bucket")
@pytest.fixture(scope="session")
def import_2pc_downstream_minio_endpoint(request):
host = (
request.config.getoption("--import-2pc-downstream-minio-host")
or request.config.getoption("--import-2pc-minio-host")
or request.config.getoption("--minio_host")
)
return _minio_endpoint(host)
@pytest.fixture(scope="session")
def import_2pc_downstream_minio_bucket(request):
return (
request.config.getoption("--import-2pc-downstream-minio-bucket")
or request.config.getoption("--import-2pc-minio-bucket")
or request.config.getoption("--minio_bucket")
)
@pytest.fixture(scope="session")
def import_2pc_rows(request):
return int(request.config.getoption("--import-2pc-rows"))
@pytest.fixture(scope="session")
def switchover_helper(request, upstream_client, downstream_client):
"""Returns a callable that performs CDC topology switchover."""
@@ -13,6 +13,7 @@ from chaos.checker import (
FullTextSearchChecker,
GeoQueryChecker,
HybridSearchChecker,
Import2PCChecker,
InsertChecker,
JsonQueryChecker,
Op,
@@ -89,7 +90,20 @@ class TestBase:
class TestOperations(TestBase):
@pytest.fixture(scope="function", autouse=True)
def connection(self, upstream_uri, upstream_token, milvus_ns):
def connection(
self,
upstream_uri,
upstream_token,
downstream_uri,
downstream_token,
milvus_ns,
import_2pc_workload,
import_2pc_minio_endpoint,
import_2pc_minio_bucket,
import_2pc_downstream_minio_endpoint,
import_2pc_downstream_minio_bucket,
import_2pc_rows,
):
connections.connect("default", uri=upstream_uri, token=upstream_token)
if connections.has_connection("default") is False:
raise Exception("no connections")
@@ -97,6 +111,19 @@ class TestOperations(TestBase):
self.milvus_sys = MilvusSys(alias="default")
self.milvus_ns = milvus_ns
self.release_name = get_milvus_instance_name(self.milvus_ns, milvus_sys=self.milvus_sys)
cf.param_info.param_uri = upstream_uri
cf.param_info.param_token = upstream_token
cf.param_info.param_bucket_name = import_2pc_minio_bucket
self.upstream_uri = upstream_uri
self.upstream_token = upstream_token
self.downstream_uri = downstream_uri
self.downstream_token = downstream_token
self.import_2pc_workload = import_2pc_workload
self.import_2pc_minio_endpoint = import_2pc_minio_endpoint
self.import_2pc_minio_bucket = import_2pc_minio_bucket
self.import_2pc_downstream_minio_endpoint = import_2pc_downstream_minio_endpoint
self.import_2pc_downstream_minio_bucket = import_2pc_downstream_minio_bucket
self.import_2pc_rows = import_2pc_rows
def init_health_checkers(self, collection_name=None):
c_name = collection_name
@@ -116,6 +143,22 @@ class TestOperations(TestBase):
Op.delete: DeleteChecker(collection_name=c_name, schema=schema),
Op.add_field: AddFieldChecker(collection_name=c_name, schema=schema),
}
if self.import_2pc_workload:
checkers[Op.import_2pc] = Import2PCChecker(
collection_name=c_name,
schema=schema,
rows_per_import=self.import_2pc_rows,
minio_endpoint=self.import_2pc_minio_endpoint,
bucket_name=self.import_2pc_minio_bucket,
uri=self.upstream_uri,
token=self.upstream_token,
downstream_uri=self.downstream_uri,
downstream_token=self.downstream_token,
downstream_minio_endpoint=self.import_2pc_downstream_minio_endpoint,
downstream_bucket_name=self.import_2pc_downstream_minio_bucket,
strict_count=False,
pk_start=-int(time.time() * 100000),
)
log.info(f"init_health_checkers: {checkers}")
self.health_checkers = checkers
@@ -18,6 +18,7 @@ from chaos.checker import (
FullTextSearchChecker,
GeoQueryChecker,
HybridSearchChecker,
Import2PCChecker,
IndexCreateChecker,
InsertChecker,
JsonQueryChecker,
@@ -58,7 +59,20 @@ class TestBase:
class TestOperations(TestBase):
@pytest.fixture(scope="function", autouse=True)
def connection(self, upstream_uri, upstream_token, milvus_ns):
def connection(
self,
upstream_uri,
upstream_token,
downstream_uri,
downstream_token,
milvus_ns,
import_2pc_workload,
import_2pc_minio_endpoint,
import_2pc_minio_bucket,
import_2pc_downstream_minio_endpoint,
import_2pc_downstream_minio_bucket,
import_2pc_rows,
):
connections.connect("default", uri=upstream_uri, token=upstream_token)
if connections.has_connection("default") is False:
raise Exception("no connections")
@@ -70,6 +84,19 @@ class TestOperations(TestBase):
self.milvus_sys = MilvusSys(alias="default")
self.milvus_ns = milvus_ns
self.release_name = get_milvus_instance_name(self.milvus_ns, milvus_sys=self.milvus_sys)
cf.param_info.param_uri = upstream_uri
cf.param_info.param_token = upstream_token
cf.param_info.param_bucket_name = import_2pc_minio_bucket
self.upstream_uri = upstream_uri
self.upstream_token = upstream_token
self.downstream_uri = downstream_uri
self.downstream_token = downstream_token
self.import_2pc_workload = import_2pc_workload
self.import_2pc_minio_endpoint = import_2pc_minio_endpoint
self.import_2pc_minio_bucket = import_2pc_minio_bucket
self.import_2pc_downstream_minio_endpoint = import_2pc_downstream_minio_endpoint
self.import_2pc_downstream_minio_bucket = import_2pc_downstream_minio_bucket
self.import_2pc_rows = import_2pc_rows
def init_health_checkers(self, collection_name=None):
c_name = collection_name
@@ -96,6 +123,19 @@ class TestOperations(TestBase):
Op.add_field: AddFieldChecker(collection_name=c_name, schema=schema),
Op.rename_collection: CollectionRenameChecker(collection_name=c_name, schema=schema),
}
if self.import_2pc_workload:
checkers[Op.import_2pc] = Import2PCChecker(
collection_name=cf.gen_unique_str("Import2PCChecker_"),
rows_per_import=self.import_2pc_rows,
minio_endpoint=self.import_2pc_minio_endpoint,
bucket_name=self.import_2pc_minio_bucket,
uri=self.upstream_uri,
token=self.upstream_token,
downstream_uri=self.downstream_uri,
downstream_token=self.downstream_token,
downstream_minio_endpoint=self.import_2pc_downstream_minio_endpoint,
downstream_bucket_name=self.import_2pc_downstream_minio_bucket,
)
self.health_checkers = checkers
@pytest.mark.tags(CaseLabel.CDC)
@@ -0,0 +1,52 @@
import pytest
from chaos.checker import Import2PCChecker
from common import common_func as cf
from common.common_type import CaseLabel
from pymilvus import connections
class TestCDCImport2PC:
@pytest.mark.tags(CaseLabel.CDC)
def test_import_2pc_manual_commit_replicates_to_downstream(
self,
upstream_client,
upstream_uri,
upstream_token,
downstream_uri,
downstream_token,
import_2pc_minio_endpoint,
import_2pc_minio_bucket,
import_2pc_downstream_minio_endpoint,
import_2pc_downstream_minio_bucket,
import_2pc_rows,
sync_timeout,
):
"""
target: verify Import 2PC manual commit is a valid CDC workload
method: create an upstream manual import job with auto_commit=false, stage the same parquet files in both
object stores, wait for primary and secondary Uncommitted, then CommitImport on upstream
expected: imported PKs stay invisible on both clusters before commit and become visible on both clusters
after the replicated CommitImport is consumed
"""
collection_name = cf.gen_unique_str("CDCImport2PC_")
connections.connect("default", uri=upstream_uri, token=upstream_token)
checker = Import2PCChecker(
collection_name=collection_name,
rows_per_import=import_2pc_rows,
minio_endpoint=import_2pc_minio_endpoint,
bucket_name=import_2pc_minio_bucket,
uri=upstream_uri,
token=upstream_token,
downstream_uri=downstream_uri,
downstream_token=downstream_token,
downstream_minio_endpoint=import_2pc_downstream_minio_endpoint,
downstream_bucket_name=import_2pc_downstream_minio_bucket,
visibility_timeout=max(sync_timeout, 180),
)
try:
res, ok = checker.run_task()
assert ok, res
finally:
checker.terminate()
if upstream_client.has_collection(collection_name):
upstream_client.drop_collection(collection_name)
+431 -5
View File
@@ -11,15 +11,18 @@ from collections import Counter
from datetime import UTC, datetime
from enum import Enum
from time import sleep
from urllib.parse import urlparse
import pandas as pd
import pytest
import requests
from chaos import constants
from common import common_func as cf
from common import common_type as ct
from common.common_type import CheckTasks
from common.milvus_sys import MilvusSys
from faker import Faker
from minio import Minio
from prettytable import PrettyTable
from pymilvus import (
AnnSearchRequest,
@@ -30,7 +33,7 @@ from pymilvus import (
RRFRanker,
connections,
)
from pymilvus.bulk_writer import BulkFileType, RemoteBulkWriter
from pymilvus.bulk_writer import BulkFileType, RemoteBulkWriter, bulk_import, get_import_progress
from pymilvus.client.embedding_list import EmbeddingList
from pymilvus.exceptions import SchemaMismatchRetryableException
from pymilvus.milvus_client.index import IndexParams
@@ -301,6 +304,7 @@ class Op(Enum):
drop_partition = "drop_partition"
load_balance = "load_balance"
bulk_insert = "bulk_insert"
import_2pc = "import_2pc"
alter_collection = "alter_collection"
add_field = "add_field"
add_vector_field = "add_vector_field"
@@ -496,12 +500,16 @@ class Checker:
self.bucket_name = cf.param_info.param_bucket_name
# Initialize MilvusClient - prioritize uri and token
if cf.param_info.param_uri:
if kwargs.get("uri"):
uri = kwargs["uri"]
elif cf.param_info.param_uri:
uri = cf.param_info.param_uri
else:
uri = "http://" + cf.param_info.param_host + ":" + str(cf.param_info.param_port)
if cf.param_info.param_token:
if kwargs.get("token"):
token = kwargs["token"]
elif cf.param_info.param_token:
token = cf.param_info.param_token
else:
token = f"{cf.param_info.param_user}:{cf.param_info.param_password}"
@@ -1890,7 +1898,6 @@ class InsertChecker(Checker):
@exception_handler()
def run_task(self):
res, result = self.insert_entities()
return res, result
@@ -2152,7 +2159,6 @@ class PartialUpdateChecker(Checker):
@exception_handler()
def run_task(self, count=0):
schema = self.get_schema()
pk_field_name = self.int64_field_name
rows = len(self.data)
@@ -3081,6 +3087,426 @@ class BulkInsertChecker(Checker):
sleep(constants.WAIT_PER_OP / 10)
class Import2PCChecker(Checker):
"""Check manual Import 2PC lifecycle in a dependent chaos thread."""
def __init__(
self,
collection_name=None,
rows_per_import=20,
dim=ct.default_dim,
schema=None,
minio_endpoint=None,
bucket_name=None,
import_timeout=720,
visibility_timeout=180,
insert_data=False,
uri=None,
token=None,
downstream_uri=None,
downstream_token=None,
downstream_minio_endpoint=None,
downstream_bucket_name=None,
strict_count=True,
pk_start=None,
):
if collection_name is None:
collection_name = cf.gen_unique_str("Import2PCChecker_")
schema = cf.gen_bulk_insert_collection_schema(dim=dim, with_json=False) if schema is None else schema
super().__init__(
collection_name=collection_name,
dim=dim,
schema=schema,
insert_data=insert_data,
uri=uri,
token=token,
)
self.schema = schema
self.rows_per_import = rows_per_import
self.import_timeout = import_timeout
self.visibility_timeout = visibility_timeout
self.minio_endpoint = minio_endpoint or "127.0.0.1:9000"
self.bucket_name = bucket_name or self.bucket_name or "milvus-bucket"
self.downstream_uri = downstream_uri
self.downstream_token = downstream_token or token or "root:Milvus"
self.downstream_minio_endpoint = downstream_minio_endpoint or self.minio_endpoint
self.downstream_bucket_name = downstream_bucket_name or self.bucket_name
self.downstream_client = (
MilvusClient(uri=self.downstream_uri, token=self.downstream_token) if self.downstream_uri else None
)
self.strict_count = strict_count
self.import_remote_path = f"import_2pc_checker/{self.c_name}"
self.pk_field_name = self.int64_field_name
if self.pk_field_name is None:
raise AssertionError("Import2PCChecker requires a schema with an INT64 primary key field")
self.uri = uri or cf.param_info.param_uri or f"http://{cf.param_info.param_host}:{cf.param_info.param_port}"
self.token = token or cf.param_info.param_token or f"{cf.param_info.param_user}:{cf.param_info.param_password}"
self.next_pk = int(pk_start) if pk_start is not None else int(time.time() * self.scale)
self.expected_count = self._query_count()
self.pending_job_ids = set()
@staticmethod
def _unwrap_data(resp_json):
data = resp_json.get("data", {})
if isinstance(data, list) and data:
return data[0]
return data or {}
@staticmethod
def _job_id_from_create_response(resp_json):
data = Import2PCChecker._unwrap_data(resp_json)
job_id = data.get("jobId") or data.get("job_id")
if isinstance(job_id, list):
job_id = job_id[0] if job_id else None
if job_id is None:
raise AssertionError(f"import create response has no job id: {resp_json}")
return str(job_id)
@staticmethod
def _job_state_from_progress(resp_json):
data = Import2PCChecker._unwrap_data(resp_json)
return data.get("state"), data.get("reason") or "", data
def _post_import_endpoint(self, endpoint, payload):
url = f"{self.uri}/v2/vectordb/jobs/import/{endpoint}"
resp = requests.post(
url,
headers={"Authorization": f"Bearer {self.token}", "Content-Type": "application/json"},
json=payload,
timeout=timeout,
)
if resp.status_code != 200:
raise AssertionError(f"import {endpoint} http status={resp.status_code}, body={resp.text}")
resp_json = resp.json()
if resp_json.get("code") != 0:
raise AssertionError(f"import {endpoint} failed: {resp_json}")
return resp_json
def _query_count(self, client=None):
client = client or self.milvus_client
res = client.query(
collection_name=self.c_name,
filter="",
output_fields=["count(*)"],
consistency_level="Strong",
timeout=query_timeout,
)
return int(res[0]["count(*)"]) if res else 0
def _query_ids(self, ids, client=None):
if not ids:
return set()
client = client or self.milvus_client
expr = f"{self.pk_field_name} in {sorted(ids)}"
res = client.query(
collection_name=self.c_name,
filter=expr,
output_fields=[self.pk_field_name],
consistency_level="Strong",
timeout=query_timeout,
)
return {row[self.pk_field_name] for row in res}
def _wait_ids_visible(self, ids, client=None, label="primary"):
deadline = time.time() + self.visibility_timeout
last_seen = set()
while time.time() < deadline:
try:
last_seen = self._query_ids(ids, client=client)
if last_seen == set(ids):
return None, True
except Exception as e:
log.debug(f"Import2PCChecker {label} query visible retry failed: {e}")
sleep(2)
return f"{label} import ids not visible, expected={sorted(ids)}, last_seen={sorted(last_seen)}", False
def _wait_ids_absent(self, ids, client=None, label="primary"):
deadline = time.time() + min(30, self.visibility_timeout)
last_seen = set()
while time.time() < deadline:
try:
last_seen = self._query_ids(ids, client=client)
if not last_seen:
return None, True
except Exception as e:
log.debug(f"Import2PCChecker {label} query absent retry failed: {e}")
sleep(2)
return f"{label} import ids became visible before commit: {sorted(last_seen)}", False
def _wait_count(self, expected, client=None, label="primary"):
deadline = time.time() + self.visibility_timeout
last_count = None
while time.time() < deadline:
try:
last_count = self._query_count(client=client)
if last_count == expected:
return None, True
except Exception as e:
log.debug(f"Import2PCChecker {label} count retry failed: {e}")
sleep(2)
return f"{label} unexpected count, expected={expected}, last_count={last_count}", False
def _wait_import_state(self, job_id, expected_state, uri=None, token=None, label="primary"):
uri = uri or self.uri
token = token or self.token
deadline = time.time() + self.import_timeout
last_progress = None
while time.time() < deadline:
resp = get_import_progress(url=uri, api_key=token, job_id=job_id, timeout=timeout)
last_progress = resp.json()
if last_progress.get("code") != 0:
sleep(2)
continue
state, reason, _ = self._job_state_from_progress(last_progress)
if state == expected_state:
return last_progress, True
if state == "Failed":
return f"{label} import job {job_id} failed while waiting for {expected_state}: {reason}", False
sleep(2)
return f"{label} import job {job_id} did not reach {expected_state}, last={last_progress}", False
def _wait_downstream_collection_ready(self):
if self.downstream_client is None:
return None, True
deadline = time.time() + self.visibility_timeout
last_error = None
while time.time() < deadline:
try:
if self.downstream_client.has_collection(self.c_name):
self._query_count(client=self.downstream_client)
return None, True
except Exception as e:
last_error = e
log.debug(f"Import2PCChecker downstream collection wait retry failed: {e}")
sleep(2)
return f"downstream collection {self.c_name} not queryable, last_error={last_error}", False
def _make_rows(self):
start = self.next_pk
self.next_pk += self.rows_per_import
rows = cf.gen_row_data_by_schema(nb=self.rows_per_import, schema=self.schema, start=start)
ids = []
for offset, row in enumerate(rows):
pk = start + offset
row[self.pk_field_name] = pk
ids.append(pk)
rows = [{key: self._normalize_import_value(value) for key, value in row.items()} for row in rows]
return rows, ids
@staticmethod
def _normalize_import_value(value):
if isinstance(value, dict):
return {key: Import2PCChecker._normalize_import_value(item) for key, item in value.items()}
if isinstance(value, (list, tuple)):
return [Import2PCChecker._normalize_import_value(item) for item in value]
if value.__class__.__module__.startswith("numpy"):
if hasattr(value, "tolist"):
return Import2PCChecker._normalize_import_value(value.tolist())
if hasattr(value, "item"):
return value.item()
return value
def _write_import_files(self, rows):
with RemoteBulkWriter(
schema=self.schema,
file_type=BulkFileType.PARQUET,
remote_path=self.import_remote_path,
connect_param=RemoteBulkWriter.ConnectParam(
endpoint=self.minio_endpoint,
access_key="minioadmin",
secret_key="minioadmin",
bucket_name=self.bucket_name,
),
) as remote_writer:
for row in rows:
remote_writer.append_row(row)
remote_writer.commit()
batch_files = remote_writer.batch_files
if not batch_files:
raise AssertionError("RemoteBulkWriter did not produce import files")
return batch_files
@staticmethod
def _flatten_import_files(files):
flattened = []
for file_item in files:
if isinstance(file_item, (list, tuple)):
flattened.extend(Import2PCChecker._flatten_import_files(file_item))
else:
flattened.append(str(file_item))
return flattened
@staticmethod
def _object_name_from_import_path(file_path, bucket_name):
parsed = urlparse(file_path)
if parsed.scheme:
path = parsed.path.lstrip("/")
if parsed.netloc == bucket_name:
return path
bucket_prefix = f"{bucket_name}/"
if path.startswith(bucket_prefix):
return path[len(bucket_prefix) :]
return path
return file_path.lstrip("/")
def _copy_import_files_to_downstream(self, files):
if self.downstream_client is None:
return []
if self.downstream_minio_endpoint == self.minio_endpoint and self.downstream_bucket_name == self.bucket_name:
return []
source = Minio(
self.minio_endpoint,
access_key="minioadmin",
secret_key="minioadmin",
secure=False,
)
target = Minio(
self.downstream_minio_endpoint,
access_key="minioadmin",
secret_key="minioadmin",
secure=False,
)
if not target.bucket_exists(self.downstream_bucket_name):
target.make_bucket(self.downstream_bucket_name)
copied = []
for file_path in self._flatten_import_files(files):
object_name = self._object_name_from_import_path(file_path, self.bucket_name)
stat = source.stat_object(self.bucket_name, object_name)
response = source.get_object(self.bucket_name, object_name)
try:
target.put_object(
self.downstream_bucket_name,
object_name,
response,
length=stat.size,
)
finally:
response.close()
response.release_conn()
copied.append(object_name)
log.info(f"Import2PCChecker copied import files to downstream object store: {copied}")
return copied
@trace()
def import_2pc(self):
job_id = None
try:
rows, ids = self._make_rows()
expected_count_after_commit = self.expected_count + len(ids)
ready_msg, downstream_ready = self._wait_downstream_collection_ready()
if not downstream_ready:
return ready_msg, False
files = self._write_import_files(rows)
self._copy_import_files_to_downstream(files)
create_resp = bulk_import(
url=self.uri,
api_key=self.token,
collection_name=self.c_name,
files=files,
options={"auto_commit": "false"},
timeout=timeout,
)
job_id = self._job_id_from_create_response(create_resp.json())
self.pending_job_ids.add(job_id)
progress, ready = self._wait_import_state(job_id, "Uncommitted")
if not ready:
return progress, False
if self.downstream_client is not None:
progress, ready = self._wait_import_state(
job_id,
"Uncommitted",
uri=self.downstream_uri,
token=self.downstream_token,
label="downstream",
)
if not ready:
return progress, False
res, absent = self._wait_ids_absent(ids)
if not absent:
return res, False
if self.downstream_client is not None:
res, absent = self._wait_ids_absent(ids, client=self.downstream_client, label="downstream")
if not absent:
return res, False
if self.strict_count:
res, count_unchanged = self._wait_count(self.expected_count)
if not count_unchanged:
return f"count changed before commit: {res}", False
if self.downstream_client is not None:
res, count_unchanged = self._wait_count(
self.expected_count, client=self.downstream_client, label="downstream"
)
if not count_unchanged:
return f"downstream count changed before commit: {res}", False
self._post_import_endpoint("commit", {"jobId": job_id})
progress, completed = self._wait_import_state(job_id, "Completed")
if not completed:
return progress, False
if self.downstream_client is not None:
progress, completed = self._wait_import_state(
job_id,
"Completed",
uri=self.downstream_uri,
token=self.downstream_token,
label="downstream",
)
if not completed:
return progress, False
res, visible = self._wait_ids_visible(ids)
if not visible:
return res, False
if self.downstream_client is not None:
res, visible = self._wait_ids_visible(ids, client=self.downstream_client, label="downstream")
if not visible:
return res, False
if self.strict_count:
res, count_ok = self._wait_count(expected_count_after_commit)
if not count_ok:
return res, False
if self.downstream_client is not None:
res, count_ok = self._wait_count(
expected_count_after_commit, client=self.downstream_client, label="downstream"
)
if not count_ok:
return res, False
self.expected_count = expected_count_after_commit
self.pending_job_ids.discard(job_id)
return {"job_id": job_id, "rows": len(ids), "expected_count": self.expected_count}, True
except Exception as e:
log.warning(f"Import2PCChecker import_2pc failed: {e}")
return str(e), False
@exception_handler()
def run_task(self):
return self.import_2pc()
def terminate(self):
self._keep_running = False
for job_id in list(self.pending_job_ids):
try:
self._post_import_endpoint("abort", {"jobId": job_id})
except Exception as e:
log.debug(f"abort pending Import2PCChecker job {job_id} failed: {e}")
finally:
self.pending_job_ids.discard(job_id)
self.reset()
def keep_running(self):
while self._keep_running:
self.run_task()
sleep(constants.WAIT_PER_OP)
class AlterCollectionChecker(Checker):
def __init__(self, collection_name=None, schema=None):
if collection_name is None:
+1 -1
View File
@@ -5,5 +5,5 @@ install milvus with authentication enabled
```bash
pip install -r requirements.txt
pytest testcases -m L0 -n 6 -v --endpoint http://127.0.0.1:19530 --minio_host 127.0.0.1
pytest testcases --tags L0 L1 -n 6 -v --endpoint http://127.0.0.1:19530 --minio_host 127.0.0.1
```
+63 -7
View File
@@ -1,3 +1,4 @@
import copy
import json
import time
import urllib.parse
@@ -450,13 +451,15 @@ class CollectionClient(Requests):
def collection_create(self, payload, db_name="default"):
time.sleep(1) # wait for collection created and in case of rate limit
payload = copy.deepcopy(payload)
c_name = payload.get("collectionName", None)
db_name = payload.get("dbName", db_name)
if self.db_name is not None:
db_name = self.db_name
elif db_name == "default":
db_name = payload.get("dbName", db_name)
self.name_list.append((db_name, c_name))
url = f"{self.endpoint}/v2/vectordb/collections/create"
if self.db_name is not None:
payload["dbName"] = self.db_name
if db_name != "default":
payload["dbName"] = db_name
if not ("params" in payload and "consistencyLevel" in payload["params"]):
@@ -563,10 +566,13 @@ class CollectionClient(Requests):
response = self.post(url, headers=self.update_headers(), data=payload)
return response.json()
def compact(self, collection_name, db_name="default"):
def compact(self, collection_name, db_name="default", is_clustering=False):
"""Compact collection"""
url = f"{self.endpoint}/v2/vectordb/collections/compact"
payload = {"collectionName": collection_name}
payload = {
"collectionName": collection_name,
"isClustering": is_clustering,
}
if self.db_name is not None:
payload["dbName"] = self.db_name
if db_name != "default":
@@ -574,10 +580,10 @@ class CollectionClient(Requests):
response = self.post(url, headers=self.update_headers(), data=payload)
return response.json()
def get_compaction_state(self, collection_name, db_name="default"):
def get_compaction_state(self, job_id, db_name="default"):
"""Get compaction state"""
url = f"{self.endpoint}/v2/vectordb/collections/get_compaction_state"
payload = {"collectionName": collection_name}
payload = {"jobID": job_id}
if self.db_name is not None:
payload["dbName"] = self.db_name
if db_name != "default":
@@ -914,6 +920,56 @@ class ImportJobClient(Requests):
res = response.json()
return res
def describe_import_job(self, job_id, db_name="default"):
if self.db_name is not None:
db_name = self.db_name
payload = {"dbName": db_name, "jobId": job_id}
if db_name is None:
payload.pop("dbName")
if job_id is None:
payload.pop("jobId")
url = f"{self.endpoint}/v2/vectordb/jobs/import/describe"
response = self.post(url, headers=self.update_headers(), data=payload)
return response.json()
def commit_import_job(self, job_id, db_name="default"):
if self.db_name is not None:
db_name = self.db_name
payload = {"dbName": db_name, "jobId": job_id}
if db_name is None:
payload.pop("dbName")
if job_id is None:
payload.pop("jobId")
url = f"{self.endpoint}/v2/vectordb/jobs/import/commit"
response = self.post(url, headers=self.update_headers(), data=payload)
return response.json()
def abort_import_job(self, job_id, db_name="default"):
if self.db_name is not None:
db_name = self.db_name
payload = {"dbName": db_name, "jobId": job_id}
if db_name is None:
payload.pop("dbName")
if job_id is None:
payload.pop("jobId")
url = f"{self.endpoint}/v2/vectordb/jobs/import/abort"
response = self.post(url, headers=self.update_headers(), data=payload)
return response.json()
def wait_import_job_state(self, job_id, expected_state, timeout=120, db_name="default", interval=2):
t0 = time.time()
last_rsp = self.get_import_job_progress(job_id, db_name=db_name)
while time.time() - t0 < timeout:
last_rsp = self.get_import_job_progress(job_id, db_name=db_name)
if last_rsp.get("code") == 0:
state = last_rsp.get("data", {}).get("state")
if state == expected_state:
return last_rsp, True
if state in ("Completed", "Failed"):
return last_rsp, False
time.sleep(interval)
return last_rsp, False
def wait_import_job_completed(self, job_id):
finished = False
t0 = time.time()
+39 -28
View File
@@ -1,12 +1,24 @@
import json
import sys
import pytest
import time
import uuid
from pymilvus import connections, db, MilvusClient
import pytest
from api.milvus import (
AliasClient,
CollectionClient,
DatabaseClient,
ImportJobClient,
IndexClient,
PartitionClient,
Requests,
RoleClient,
StorageClient,
UserClient,
VectorClient,
)
from pymilvus import connections, db
from utils.util_log import test_log as logger
from api.milvus import (VectorClient, CollectionClient, PartitionClient, IndexClient, AliasClient,
UserClient, RoleClient, ImportJobClient, StorageClient, Requests, DatabaseClient)
from utils.utils import get_data_by_payload
@@ -42,39 +54,35 @@ class TestBase(Base):
def teardown_method(self):
# Clean up collections
if hasattr(self, 'api_key') and self.api_key:
if hasattr(self, "api_key") and self.api_key:
self.collection_client.api_key = self.api_key
all_collections = self.collection_client.collection_list()['data']
all_collections = self.collection_client.collection_list()["data"]
if self.name in all_collections:
logger.info(f"collection {self.name} exist, drop it")
payload = {
"collectionName": self.name,
}
try:
rsp = self.collection_client.collection_drop(payload)
self.collection_client.collection_drop(payload)
except Exception as e:
logger.error(f"drop collection error: {e}")
for item in self.collection_client.name_list:
db_name = item[0]
c_name = item[1]
payload = {
"collectionName": c_name,
"dbName": db_name
}
payload = {"collectionName": c_name, "dbName": db_name}
try:
self.collection_client.collection_drop(payload)
except Exception as e:
logger.error(f"drop collection error: {e}")
# Clean up databases created by this client
if hasattr(self, 'api_key') and self.api_key:
if hasattr(self, "api_key") and self.api_key:
self.database_client.api_key = self.api_key
for db_name in self.database_client.db_names[:]: # Create a copy of the list to iterate
logger.info(f"database {db_name} exist, drop it")
try:
rsp = self.database_client.database_drop({"dbName": db_name})
self.database_client.database_drop({"dbName": db_name})
except Exception as e:
logger.error(f"drop database error: {e}")
@@ -105,7 +113,16 @@ class TestBase(Base):
self.partition_client.api_key = None
connections.connect(uri=endpoint, token=token)
def init_collection(self, collection_name, pk_field="id", metric_type="L2", dim=128, nb=3000, batch_size=1000, return_insert_id=False):
def init_collection(
self,
collection_name,
pk_field="id",
metric_type="L2",
dim=128,
nb=3000,
batch_size=1000,
return_insert_id=False,
):
# create collection
schema_payload = {
"collectionName": collection_name,
@@ -116,7 +133,7 @@ class TestBase(Base):
"vectorField": "vector",
}
rsp = self.collection_client.collection_create(schema_payload)
assert rsp['code'] == 0
assert rsp["code"] == 0
self.wait_collection_load_completed(collection_name)
batch_size = batch_size
batch = nb // batch_size
@@ -127,29 +144,23 @@ class TestBase(Base):
for i in range(batch):
nb = batch_size
data = get_data_by_payload(schema_payload, nb)
payload = {
"collectionName": collection_name,
"data": data
}
payload = {"collectionName": collection_name, "data": data}
body_size = sys.getsizeof(json.dumps(payload))
logger.debug(f"body size: {body_size / 1024 / 1024} MB")
rsp = self.vector_client.vector_insert(payload)
assert rsp['code'] == 0
assert rsp["code"] == 0
if return_insert_id:
insert_ids.extend(rsp['data']['insertIds'])
insert_ids.extend(rsp["data"]["insertIds"])
full_data.extend(data)
# insert remainder data
if remainder:
nb = remainder
data = get_data_by_payload(schema_payload, nb)
payload = {
"collectionName": collection_name,
"data": data
}
payload = {"collectionName": collection_name, "data": data}
rsp = self.vector_client.vector_insert(payload)
assert rsp['code'] == 0
assert rsp["code"] == 0
if return_insert_id:
insert_ids.extend(rsp['data']['insertIds'])
insert_ids.extend(rsp["data"]["insertIds"])
full_data.extend(data)
if return_insert_id:
return schema_payload, full_data, insert_ids
+10 -10
View File
@@ -11,32 +11,32 @@ class LogConfig:
@staticmethod
def get_env_variable(var="CI_LOG_PATH"):
""" get log path for testing """
"""get log path for testing"""
try:
log_path = os.environ[var]
return str(log_path)
except Exception as e:
# now = datetime.datetime.now().strftime("%Y-%m-%d-%H-%M-%S")
log_path = f"/tmp/ci_logs"
print("[get_env_variable] failed to get environment variables : %s, use default path : %s" % (str(e), log_path))
log_path = "/tmp/ci_logs"
print(f"[get_env_variable] failed to get environment variables : {e}, use default path : {log_path}")
return log_path
@staticmethod
def create_path(log_path):
if not os.path.isdir(str(log_path)):
print("[create_path] folder(%s) is not exist." % log_path)
print(f"[create_path] folder({log_path}) is not exist.")
print("[create_path] create path now...")
os.makedirs(log_path)
def get_default_config(self):
""" Make sure the path exists """
"""Make sure the path exists"""
log_dir = self.get_env_variable()
self.log_debug = "%s/ci_test_log.debug" % log_dir
self.log_info = "%s/ci_test_log.log" % log_dir
self.log_err = "%s/ci_test_log.err" % log_dir
work_log = os.environ.get('PYTEST_XDIST_WORKER')
self.log_debug = f"{log_dir}/ci_test_log.debug"
self.log_info = f"{log_dir}/ci_test_log.log"
self.log_err = f"{log_dir}/ci_test_log.err"
work_log = os.environ.get("PYTEST_XDIST_WORKER")
if work_log is not None:
self.log_worker = f'{log_dir}/{work_log}.log'
self.log_worker = f"{log_dir}/{work_log}.log"
self.create_path(log_dir)
+76 -3
View File
@@ -9,9 +9,35 @@ def pytest_addoption(parser):
parser.addoption("--bucket_name", action="store", default="milvus-bucket", help="minio bucket name")
parser.addoption("--root_path", action="store", default="file", help="minio bucket root path")
parser.addoption("--release_name", action="store", default="my-release", help="release name")
parser.addoption("--secondary_endpoint", action="store", default=None, help="secondary Milvus endpoint")
parser.addoption("--secondary_token", action="store", default="root:Milvus", help="secondary Milvus token")
parser.addoption("--secondary_release_name", action="store", default=None, help="secondary Milvus release name")
parser.addoption("--secondary_minio_host", action="store", default=None, help="secondary MinIO host")
parser.addoption("--secondary_bucket_name", action="store", default=None, help="secondary MinIO bucket name")
parser.addoption("--secondary_root_path", action="store", default="file", help="secondary MinIO root path")
parser.addoption(
"--source-cluster-id", "--source_cluster_id", action="store", default=None, help="CDC source cluster ID"
)
parser.addoption(
"--target-cluster-id", "--target_cluster_id", action="store", default=None, help="CDC target cluster ID"
)
parser.addoption(
"--pchannel-num", "--pchannel_num", action="store", default="16", help="CDC physical channel count"
)
# a tei endpoint for text embedding, default is http://text-embeddings-service.milvus-ci.svc.cluster.local:80 which is deployed in house
parser.addoption("--tei_endpoint", action="store", default="http://text-embeddings-service.milvus-ci.svc.cluster.local:80", help="tei endpoint")
parser.addoption("--tei_reranker_endpoint", action="store", default="http://text-rerank-service.milvus-ci.svc.cluster.local:80", help="tei reranker endpoint")
parser.addoption(
"--tei_endpoint",
action="store",
default="http://text-embeddings-service.milvus-ci.svc.cluster.local:80",
help="tei endpoint",
)
parser.addoption(
"--tei_reranker_endpoint",
action="store",
default="http://text-rerank-service.milvus-ci.svc.cluster.local:80",
help="tei reranker endpoint",
)
@pytest.fixture
def endpoint(request):
@@ -42,10 +68,57 @@ def root_path(request):
def release_name(request):
return request.config.getoption("--release_name")
@pytest.fixture
def secondary_endpoint(request):
return request.config.getoption("--secondary_endpoint")
@pytest.fixture
def secondary_token(request):
return request.config.getoption("--secondary_token")
@pytest.fixture
def secondary_release_name(request):
return request.config.getoption("--secondary_release_name")
@pytest.fixture
def secondary_minio_host(request):
return request.config.getoption("--secondary_minio_host")
@pytest.fixture
def secondary_bucket_name(request):
return request.config.getoption("--secondary_bucket_name")
@pytest.fixture
def secondary_root_path(request):
return request.config.getoption("--secondary_root_path")
@pytest.fixture
def source_cluster_id(request):
return request.config.getoption("--source_cluster_id")
@pytest.fixture
def target_cluster_id(request):
return request.config.getoption("--target_cluster_id")
@pytest.fixture
def pchannel_num(request):
return int(request.config.getoption("--pchannel_num"))
@pytest.fixture
def tei_endpoint(request):
return request.config.getoption("--tei_endpoint")
@pytest.fixture
def tei_reranker_endpoint(request):
return request.config.getoption("--tei_reranker_endpoint")
return request.config.getoption("--tei_reranker_endpoint")
+1 -5
View File
@@ -9,10 +9,6 @@ filterwarnings =
ignore::DeprecationWarning
markers =
L0 : 'L0 case, high priority'
L1 : 'L1 case, second priority'
L2 : 'L2 case, system level case'
BulkInsert : 'Bulk Insert case'
ExternalCollection : 'External collection case'
tags: custom tags for test cases
timeout_method = thread
+1
View File
@@ -2,6 +2,7 @@
requests>=2.32.4
urllib3>=2.5.0
pytest==8.3.4
pytest_tagging==1.6.0
PyYAML==6.0.2
numpy==2.2.6
allure-pytest>=2.8.18
@@ -1,18 +1,18 @@
import random
from sklearn import preprocessing
import numpy as np
from utils.utils import gen_collection_name
from utils.util_log import test_log as logger
import pytest
from base.testbase import TestBase
from sklearn import preprocessing
from utils.constant import CaseLabel
from utils.util_log import test_log as logger
from utils.utils import gen_collection_name
@pytest.mark.L0
@pytest.mark.tags(CaseLabel.L0)
class TestAliasE2E(TestBase):
def test_alias_e2e(self):
"""
"""
""" """
# list alias before create
rsp = self.alias_client.list_alias()
name = gen_collection_name()
@@ -24,28 +24,25 @@ class TestAliasE2E(TestBase):
{"fieldName": "book_id", "dataType": "Int64", "isPrimary": True, "elementTypeParams": {}},
{"fieldName": "word_count", "dataType": "Int64", "elementTypeParams": {}},
{"fieldName": "book_describe", "dataType": "VarChar", "elementTypeParams": {"max_length": "256"}},
{"fieldName": "book_intro", "dataType": "FloatVector", "elementTypeParams": {"dim": f"{128}"}}
{"fieldName": "book_intro", "dataType": "FloatVector", "elementTypeParams": {"dim": f"{128}"}},
]
},
"indexParams": [{"fieldName": "book_intro", "indexName": "book_intro_vector", "metricType": "L2"}]
"indexParams": [{"fieldName": "book_intro", "indexName": "book_intro_vector", "metricType": "L2"}],
}
logger.info(f"create collection {name} with payload: {payload}")
rsp = client.collection_create(payload)
# create alias
alias_name = name + "_alias"
payload = {
"collectionName": name,
"aliasName": alias_name
}
payload = {"collectionName": name, "aliasName": alias_name}
rsp = self.alias_client.create_alias(payload)
assert rsp['code'] == 0
assert rsp["code"] == 0
# list alias after create
rsp = self.alias_client.list_alias()
assert alias_name in rsp['data']
assert alias_name in rsp["data"]
# describe alias
rsp = self.alias_client.describe_alias(alias_name)
assert rsp['data']["aliasName"] == alias_name
assert rsp['data']["collectionName"] == name
assert rsp["data"]["aliasName"] == alias_name
assert rsp["data"]["collectionName"] == name
# do crud operation by alias
# insert data by alias
@@ -59,16 +56,10 @@ class TestAliasE2E(TestBase):
}
data.append(tmp)
payload = {
"collectionName": alias_name,
"data": data
}
payload = {"collectionName": alias_name, "data": data}
rsp = self.vector_client.vector_insert(payload)
# delete data by alias
payload = {
"collectionName": alias_name,
"ids": [1, 2, 3]
}
payload = {"collectionName": alias_name, "ids": [1, 2, 3]}
rsp = self.vector_client.vector_delete(payload)
# upsert data by alias
@@ -81,22 +72,16 @@ class TestAliasE2E(TestBase):
"book_intro": preprocessing.normalize([np.array([random.random() for _ in range(128)])])[0].tolist(),
}
upsert_data.append(tmp)
payload = {
"collectionName": alias_name,
"data": upsert_data
}
payload = {"collectionName": alias_name, "data": upsert_data}
rsp = self.vector_client.vector_upsert(payload)
# search data by alias
payload = {
"collectionName": alias_name,
"vector": preprocessing.normalize([np.array([random.random() for i in range(128)])])[0].tolist()
"vector": preprocessing.normalize([np.array([random.random() for i in range(128)])])[0].tolist(),
}
rsp = self.vector_client.vector_search(payload)
# query data by alias
payload = {
"collectionName": alias_name,
"filter": "book_id > 10"
}
payload = {"collectionName": alias_name, "filter": "book_id > 10"}
rsp = self.vector_client.vector_query(payload)
# alter alias to another collection
@@ -107,19 +92,13 @@ class TestAliasE2E(TestBase):
"dimension": 128,
}
rsp = client.collection_create(payload)
payload = {
"collectionName": new_name,
"aliasName": alias_name
}
payload = {"collectionName": new_name, "aliasName": alias_name}
rsp = self.alias_client.alter_alias(payload)
# describe alias
rsp = self.alias_client.describe_alias(alias_name)
assert rsp['data']["aliasName"] == alias_name
assert rsp['data']["collectionName"] == new_name
assert rsp["data"]["aliasName"] == alias_name
assert rsp["data"]["collectionName"] == new_name
# query data by alias, expect no data
payload = {
"collectionName": alias_name,
"filter": "id > 0"
}
payload = {"collectionName": alias_name, "filter": "id > 0"}
rsp = self.vector_client.vector_query(payload)
assert rsp['data'] == []
assert rsp["data"] == []
@@ -9,12 +9,12 @@ import pytest
from api.milvus import CollectionClient
from base.testbase import TestBase
from pymilvus import Collection, CollectionSchema, DataType, FieldSchema
from utils.constant import default_nb
from utils.constant import CaseLabel, default_nb
from utils.util_log import test_log as logger
from utils.utils import gen_collection_name, gen_vector, get_data_by_payload
@pytest.mark.L0
@pytest.mark.tags(CaseLabel.L0)
class TestCreateCollection(TestBase):
@pytest.mark.parametrize("dim", [128])
def test_create_collections_quick_setup(self, dim):
@@ -428,8 +428,8 @@ class TestCreateCollection(TestBase):
def test_create_collections_multi_float_vector_with_one_index(self, dim, metric_type):
"""
target: test create collection
method: create a collection with a simple schema
expected: create collection success
method: create a collection with multiple vector fields and only one index
expected: create reports the missing vector index and the collection stays unloaded
"""
name = gen_collection_name()
dim = 128
@@ -452,6 +452,7 @@ class TestCreateCollection(TestBase):
logging.info(f"create collection {name} with payload: {payload}")
rsp = client.collection_create(payload)
assert rsp["code"] == 1100
assert "there is no vector index on field: [image_intro]" in rsp["message"]
rsp = client.collection_list()
all_collections = rsp["data"]
@@ -765,7 +766,7 @@ class TestCreateCollection(TestBase):
assert rsp["data"]["fields"][2]["nullable"] is True
@pytest.mark.L1
@pytest.mark.tags(CaseLabel.L1)
class TestCreateCollectionNegative(TestBase):
def test_create_collections_custom_with_invalid_datatype(self):
"""
@@ -940,7 +941,7 @@ class TestCreateCollectionNegative(TestBase):
assert "convert defaultValue fail" in rsp["message"]
@pytest.mark.L0
@pytest.mark.tags(CaseLabel.L0)
class TestHasCollections(TestBase):
def test_has_collections_default(self):
"""
@@ -988,7 +989,7 @@ class TestHasCollections(TestBase):
assert rsp["data"]["has"] is False
@pytest.mark.L0
@pytest.mark.tags(CaseLabel.L0)
class TestGetCollectionStats(TestBase):
def test_get_collections_stats(self):
"""
@@ -1024,7 +1025,7 @@ class TestGetCollectionStats(TestBase):
assert rsp["data"]["rowCount"] == nb
@pytest.mark.L0
@pytest.mark.tags(CaseLabel.L0)
class TestLoadReleaseCollection(TestBase):
def test_load_and_release_collection(self):
name = gen_collection_name()
@@ -1069,7 +1070,7 @@ class TestLoadReleaseCollection(TestBase):
assert rsp["data"]["loadState"] == "LoadStateNotLoad"
@pytest.mark.L0
@pytest.mark.tags(CaseLabel.L0)
class TestGetCollectionLoadState(TestBase):
def test_get_collection_load_state(self):
"""
@@ -1110,7 +1111,7 @@ class TestGetCollectionLoadState(TestBase):
assert rsp["data"]["loadState"] == "LoadStateLoaded"
@pytest.mark.L0
@pytest.mark.tags(CaseLabel.L0)
class TestListCollections(TestBase):
def test_list_collections_default(self):
"""
@@ -1138,7 +1139,7 @@ class TestListCollections(TestBase):
assert name in all_collections
@pytest.mark.L0
@pytest.mark.tags(CaseLabel.L0)
class TestDescribeCollection(TestBase):
def test_describe_collections_default(self):
"""
@@ -1210,7 +1211,7 @@ class TestDescribeCollection(TestBase):
assert rsp["data"]["enableDynamicField"] is True
@pytest.mark.L0
@pytest.mark.tags(CaseLabel.L0)
class TestDescribeCollectionNegative(TestBase):
def test_describe_collections_with_invalid_collection_name(self):
"""
@@ -1237,7 +1238,7 @@ class TestDescribeCollectionNegative(TestBase):
assert "can't find collection" in rsp["message"]
@pytest.mark.L0
@pytest.mark.tags(CaseLabel.L0)
class TestDropCollection(TestBase):
def test_drop_collections_default(self):
"""
@@ -1271,7 +1272,7 @@ class TestDropCollection(TestBase):
assert name not in all_collections
@pytest.mark.L0
@pytest.mark.tags(CaseLabel.L0)
class TestDropCollectionNegative(TestBase):
def test_drop_collections_with_invalid_collection_name(self):
"""
@@ -1300,7 +1301,7 @@ class TestDropCollectionNegative(TestBase):
assert rsp["code"] == 0
@pytest.mark.L0
@pytest.mark.tags(CaseLabel.L0)
class TestRenameCollection(TestBase):
def test_rename_collection(self):
"""
@@ -1334,7 +1335,7 @@ class TestRenameCollection(TestBase):
assert name not in all_collections
@pytest.mark.L1
@pytest.mark.tags(CaseLabel.RBAC)
class TestCollectionWithAuth(TestBase):
def test_drop_collections_with_invalid_api_key(self):
"""
@@ -1431,7 +1432,7 @@ class TestCollectionWithAuth(TestBase):
assert rsp["code"] == 1800
@pytest.mark.L0
@pytest.mark.tags(CaseLabel.L0)
class TestCollectionProperties(TestBase):
"""Test collection property operations"""
@@ -1580,6 +1581,7 @@ class TestCollectionProperties(TestBase):
assert p["value"] == "100"
@pytest.mark.tags(CaseLabel.L1)
class TestCollectionAddField(TestBase):
"""Test collection add field operations"""
@@ -1918,7 +1920,7 @@ class TestCollectionAddField(TestBase):
np.testing.assert_allclose(actual["sub_vec"], expected["sub_vec"], rtol=1e-5, atol=1e-5)
@pytest.mark.L1
@pytest.mark.tags(CaseLabel.L1)
class TestCollectionAddFieldNegative(TestBase):
"""Test collection add field negative cases"""
@@ -2106,7 +2108,7 @@ class TestCollectionAddFieldNegative(TestBase):
assert "collection" in rsp.get("message", "").lower() or "not found" in rsp.get("message", "").lower()
@pytest.mark.L0
@pytest.mark.tags(CaseLabel.L0)
class TestCollectionMaintenance(TestBase):
"""Test collection maintenance operations"""
@@ -2157,7 +2159,7 @@ class TestCollectionMaintenance(TestBase):
def test_collection_compact(self):
"""
target: test collection compact
method: create collection, insert data, flush multiple times, then compact
method: create collection with a clustering key, insert data, flush multiple times, then compact
expected: compact successfully
"""
# Create collection
@@ -2169,6 +2171,12 @@ class TestCollectionMaintenance(TestBase):
"schema": {
"fields": [
{"fieldName": "book_id", "dataType": "Int64", "isPrimary": True, "elementTypeParams": {}},
{
"fieldName": "word_count",
"dataType": "Int64",
"isClusteringKey": True,
"elementTypeParams": {},
},
{"fieldName": "my_vector", "dataType": "FloatVector", "elementTypeParams": {"dim": 128}},
]
},
@@ -2176,13 +2184,20 @@ class TestCollectionMaintenance(TestBase):
client.collection_create(payload)
# Insert and flush multiple times
for i in range(3):
batch_count = 4
rows_per_batch = 100
for batch in range(batch_count):
# Insert data
vectors = [gen_vector(dim=128) for _ in range(10)]
vectors = [gen_vector(dim=128) for _ in range(rows_per_batch)]
insert_data = {
"collectionName": name,
"data": [
{"book_id": i * 10 + j, "my_vector": vector} for i, vector in enumerate(vectors) for j in range(10)
{
"book_id": batch * rows_per_batch + i,
"word_count": batch,
"my_vector": vector,
}
for i, vector in enumerate(vectors)
],
}
response = vector_client.vector_insert(insert_data)
@@ -2192,11 +2207,13 @@ class TestCollectionMaintenance(TestBase):
c = Collection(name)
c.flush()
# Compact collection
response = client.compact(name)
response = client.compact(name, is_clustering=True)
assert response["code"] == 0
compaction_id = response.get("data", {}).get("compactionID")
assert isinstance(compaction_id, int) and compaction_id > 0, response
# Get compaction state
response = client.get_compaction_state(name)
response = client.get_compaction_state(compaction_id)
assert response["code"] == 0
assert "state" in response["data"]
assert "compactionID" in response["data"]
@@ -2285,7 +2302,7 @@ def _gen_struct_array_row(row_id, num_elems, dim=DEFAULT_STRUCT_ARRAY_DIM):
}
@pytest.mark.L0
@pytest.mark.tags(CaseLabel.L0)
class TestStructArrayCollection(TestBase):
def test_create_struct_array_collection(self):
name = gen_collection_name()
@@ -2312,7 +2329,7 @@ class TestStructArrayCollection(TestBase):
assert by_name["sub_vec"]["elementType"] == "FloatVector"
@pytest.mark.L1
@pytest.mark.tags(CaseLabel.L1)
class TestStructArraySchemaValidation(TestBase):
def _create_with_bad_sub_field(self, name, bad_sub_field):
payload = {
@@ -2487,7 +2504,7 @@ class TestStructArraySchemaValidation(TestBase):
assert rsp["code"] != 0, rsp
@pytest.mark.L0
@pytest.mark.tags(CaseLabel.L0)
class TestStructArrayInsertQuery(TestBase):
def _create_and_load(self, name, dim=DEFAULT_STRUCT_ARRAY_DIM):
payload = _build_struct_array_schema_payload(name, dim=dim, include_index_params=True)
@@ -2555,7 +2572,7 @@ class TestStructArrayInsertQuery(TestBase):
assert rsp["code"] != 0, rsp
@pytest.mark.L0
@pytest.mark.tags(CaseLabel.L0)
class TestStructSubVectorSearch(TestBase):
def _setup_collection_with_sub_index(self, name, dim=DEFAULT_STRUCT_ARRAY_DIM, nb=50, sub_metric="COSINE"):
payload = _build_struct_array_schema_payload(name, dim=dim, include_index_params=False)
@@ -2662,7 +2679,7 @@ class TestStructSubVectorSearch(TestBase):
assert rsp["code"] != 0, rsp
@pytest.mark.L0
@pytest.mark.tags(CaseLabel.L0)
class TestStructSubVectorSearchOneStep(TestBase):
def _create_load_insert(self, name, sub_metric, nb=50):
payload = _build_struct_array_schema_payload(
@@ -1,9 +1,10 @@
import pytest
from base.testbase import TestBase
from utils.constant import CaseLabel
from utils.utils import gen_unique_str
@pytest.mark.L0
@pytest.mark.tags(CaseLabel.L0)
class TestDatabaseOperation(TestBase):
"""
Test cases for database operations
@@ -36,8 +37,7 @@ class TestDatabaseOperation(TestBase):
describe_rsp = self.database_client.database_describe({"dbName": db_name})
assert describe_rsp["code"] == 0
assert any(
prop["key"] == "mmap.enabled" and prop["value"] == "true"
for prop in describe_rsp["data"]["properties"]
prop["key"] == "mmap.enabled" and prop["value"] == "true" for prop in describe_rsp["data"]["properties"]
)
def test_alter_database_properties(self):
@@ -54,8 +54,7 @@ class TestDatabaseOperation(TestBase):
describe_rsp = self.database_client.database_describe({"dbName": db_name})
assert describe_rsp["code"] == 0
assert any(
prop["key"] == "mmap.enabled" and prop["value"] == "true"
for prop in describe_rsp["data"]["properties"]
prop["key"] == "mmap.enabled" and prop["value"] == "true" for prop in describe_rsp["data"]["properties"]
)
# Alter properties
@@ -67,8 +66,7 @@ class TestDatabaseOperation(TestBase):
describe_rsp = self.database_client.database_describe({"dbName": db_name})
assert describe_rsp["code"] == 0
assert any(
prop["key"] == "mmap.enabled" and prop["value"] == "false"
for prop in describe_rsp["data"]["properties"]
prop["key"] == "mmap.enabled" and prop["value"] == "false" for prop in describe_rsp["data"]["properties"]
)
def test_list_databases(self):
@@ -93,9 +91,7 @@ class TestDatabaseOperation(TestBase):
properties = {"mmap.enabled": True}
# Create database
self.database_client.database_create(
{"dbName": db_name, "properties": properties}
)
self.database_client.database_create({"dbName": db_name, "properties": properties})
# Describe database
rsp = self.database_client.database_describe({"dbName": db_name})
@@ -105,7 +101,7 @@ class TestDatabaseOperation(TestBase):
assert len(rsp["data"]["properties"]) > 0
@pytest.mark.L0
@pytest.mark.tags(CaseLabel.L0)
class TestDatabaseOperationNegative(TestBase):
"""
Negative test cases for database operations
@@ -164,11 +160,10 @@ class TestDatabaseOperationNegative(TestBase):
assert rsp["code"] != 0
@pytest.mark.L0
@pytest.mark.tags(CaseLabel.L0)
class TestDatabaseProperties(TestBase):
"""Test database properties operations"""
def test_alter_database_properties(self):
"""
target: test alter database properties
@@ -178,9 +173,7 @@ class TestDatabaseProperties(TestBase):
# Create database
client = self.database_client
db_name = "test_alter_props"
payload = {
"dbName": db_name
}
payload = {"dbName": db_name}
response = client.database_create(payload)
assert response["code"] == 0
orders = [[True, False], [False, True]]
File diff suppressed because it is too large Load Diff
@@ -19,6 +19,7 @@ from pyiceberg.catalog.sql import SqlCatalog
from pyiceberg.schema import Schema as IcebergSchema
from pyiceberg.types import FixedType, FloatType, LongType, NestedField
from tenacity import Retrying, retry_if_result, stop_after_delay, wait_fixed
from utils.constant import CaseLabel
from utils.utils import gen_collection_name
DIM = 8
@@ -873,7 +874,6 @@ def _assert_basic_row_body(row, row_id, include_embedding=True):
_assert_vector_close(row["embedding"], _expected_embedding(row_id))
@pytest.mark.ExternalCollection
class TestRestExternalCollection(TestBase):
def _external_job_post(self, action, payload):
url = f"{self.endpoint}/v2/vectordb/jobs/external_collection/{action}"
@@ -1070,7 +1070,7 @@ class TestRestExternalCollection(TestBase):
all_records = _assert_job_list_response(all_jobs)
assert job_id in [job["jobId"] for job in all_records]
@pytest.mark.L0
@pytest.mark.tags(CaseLabel.L0)
def test_rest_external_collection_create_describe_metadata_parquet(self, external_store):
"""
target: verify REST v2 external collection metadata for parquet
@@ -1091,7 +1091,7 @@ class TestRestExternalCollection(TestBase):
assert fields["value"]["externalField"] == "value"
assert fields["embedding"]["externalField"] == "embedding"
@pytest.mark.L1
@pytest.mark.tags(CaseLabel.L1)
@pytest.mark.parametrize("fmt", FORMAT_CASES, ids=FORMAT_IDS)
def test_rest_external_collection_refresh_query_by_format(self, fmt, external_store):
"""
@@ -1136,7 +1136,7 @@ class TestRestExternalCollection(TestBase):
if "distance" in hit:
assert isinstance(hit["distance"], (int, float)), hit
@pytest.mark.L1
@pytest.mark.tags(CaseLabel.L1)
def test_rest_external_collection_refresh_override_persists_and_reuses(self, external_store):
"""
target: verify refresh override source/spec semantics
@@ -1180,7 +1180,7 @@ class TestRestExternalCollection(TestBase):
self._load_and_wait(name)
assert self._query_count(name, "id >= 1000") == 20
@pytest.mark.L1
@pytest.mark.tags(CaseLabel.L1)
def test_rest_external_collection_create_rejections(self, external_store):
"""
target: verify REST v2 create-time external collection validation
@@ -1293,7 +1293,7 @@ class TestRestExternalCollection(TestBase):
rsp = self.collection_client.collection_create(copy.deepcopy(payload))
_assert_error_response(rsp, message)
@pytest.mark.L1
@pytest.mark.tags(CaseLabel.L1)
def test_rest_external_collection_refresh_rejections(self, external_store):
"""
target: verify REST v2 external refresh/job API validation
@@ -1386,7 +1386,7 @@ class TestRestExternalCollection(TestBase):
rsp = self._describe_external_collection_job(9223372036854775807)
_assert_error_response(rsp, "not found")
@pytest.mark.L1
@pytest.mark.tags(CaseLabel.L1)
def test_rest_external_collection_raw_request_body_validation(self):
"""
target: verify REST v2 external job raw body validation
@@ -1402,7 +1402,7 @@ class TestRestExternalCollection(TestBase):
rsp = self._external_job_raw_post(action, body)
_assert_error_response(rsp, message)
@pytest.mark.L1
@pytest.mark.tags(CaseLabel.L1)
@pytest.mark.parametrize("fmt", FORMAT_CASES, ids=FORMAT_IDS)
def test_rest_external_collection_zero_row_refresh_boundary_by_format(self, fmt, external_store):
"""
@@ -1431,7 +1431,7 @@ class TestRestExternalCollection(TestBase):
assert progress["data"]["reason"], progress
self._assert_job_list_contains(name, job_id)
@pytest.mark.L1
@pytest.mark.tags(CaseLabel.L1)
@pytest.mark.parametrize("fmt", FORMAT_CASES, ids=FORMAT_IDS)
def test_rest_external_collection_nullable_scalar_by_format(self, fmt, external_store):
"""
@@ -1460,7 +1460,7 @@ class TestRestExternalCollection(TestBase):
_assert_vector_close(rows_by_id[1]["embedding"], _expected_embedding(1))
_assert_basic_row_body(rows_by_id[2], 2)
@pytest.mark.L1
@pytest.mark.tags(CaseLabel.L1)
@pytest.mark.parametrize("fmt", FORMAT_CASES, ids=FORMAT_IDS)
def test_rest_external_collection_nullable_vector_by_format(self, fmt, external_store):
"""
@@ -1498,7 +1498,7 @@ class TestRestExternalCollection(TestBase):
_assert_close(rows_by_id[1]["value"], _expected_value(1), tolerance=1e-4)
_assert_basic_row_body(rows_by_id[2], 2)
@pytest.mark.L1
@pytest.mark.tags(CaseLabel.L1)
def test_rest_external_collection_custom_db_e2e(self, external_store):
"""
target: verify REST v2 external collection lifecycle in a custom database
@@ -1531,7 +1531,7 @@ class TestRestExternalCollection(TestBase):
assert len(rows) == 1, rows
_assert_basic_row_body(rows[0], 0)
@pytest.mark.L1
@pytest.mark.tags(CaseLabel.RBAC)
def test_rest_external_collection_job_api_invalid_token(self):
"""
target: verify external job APIs reject invalid tokens when auth is enforced
@@ -1550,7 +1550,7 @@ class TestRestExternalCollection(TestBase):
assert rsp["code"] == 1800, f"{action} did not reject invalid token: {rsp}"
assert "message" in rsp and rsp["message"], rsp
@pytest.mark.L1
@pytest.mark.tags(CaseLabel.L1)
def test_rest_external_collection_duplicate_refresh_rejected_or_reuses_job(self, external_store):
"""
target: verify duplicate refresh behavior while a job is in progress
@@ -1582,7 +1582,7 @@ class TestRestExternalCollection(TestBase):
_assert_job_info_response(progress, expected_collection=name, expected_job_id=first_job_id)
self._assert_job_list_contains(name, first_job_id)
@pytest.mark.L1
@pytest.mark.tags(CaseLabel.L1)
def test_rest_external_collection_job_list_includes_multiple_refreshes(self, external_store):
"""
target: verify REST v2 external job list includes multiple refresh jobs
@@ -1604,7 +1604,7 @@ class TestRestExternalCollection(TestBase):
assert first_job_id in listed_ids, records
assert second_job_id in listed_ids, records
@pytest.mark.L1
@pytest.mark.tags(CaseLabel.L1)
def test_rest_external_collection_add_field_external_mapping_rejected(self):
"""
target: verify external field mappings cannot be added to regular collections
@@ -1,17 +1,17 @@
import random
import pytest
import numpy as np
from sklearn import preprocessing
from base.testbase import TestBase
from utils.utils import gen_collection_name, generate_wkt_by_type
from utils.util_log import test_log as logger
from utils.constant import default_nb
import numpy as np
import pytest
from base.testbase import TestBase
from sklearn import preprocessing
from utils.constant import CaseLabel, default_nb
from utils.util_log import test_log as logger
from utils.utils import gen_collection_name, generate_wkt_by_type
default_dim = 128
@pytest.mark.L0
@pytest.mark.tags(CaseLabel.L0)
class TestGeometryCollection(TestBase):
"""Test geometry collection operations"""
@@ -30,29 +30,22 @@ class TestGeometryCollection(TestBase):
"fields": [
{"fieldName": "id", "dataType": "Int64", "isPrimary": True},
{"fieldName": "vector", "dataType": "FloatVector", "elementTypeParams": {"dim": f"{default_dim}"}},
{"fieldName": "geo", "dataType": "Geometry"}
]
{"fieldName": "geo", "dataType": "Geometry"},
],
},
"indexParams": [
{"fieldName": "vector", "indexName": "vector_idx", "metricType": "L2"}
]
"indexParams": [{"fieldName": "vector", "indexName": "vector_idx", "metricType": "L2"}],
}
rsp = self.collection_client.collection_create(payload)
assert rsp['code'] == 0
assert rsp["code"] == 0
# Verify collection exists
rsp = self.collection_client.collection_describe(name)
assert rsp['code'] == 0
assert rsp["code"] == 0
logger.info(f"Collection created: {rsp}")
@pytest.mark.parametrize("wkt_type", [
"POINT",
"LINESTRING",
"POLYGON",
"MULTIPOINT",
"MULTILINESTRING",
"MULTIPOLYGON",
"GEOMETRYCOLLECTION"
])
@pytest.mark.parametrize(
"wkt_type",
["POINT", "LINESTRING", "POLYGON", "MULTIPOINT", "MULTILINESTRING", "MULTIPOLYGON", "GEOMETRYCOLLECTION"],
)
def test_insert_wkt_data(self, wkt_type):
"""
target: test insert various WKT geometry types
@@ -68,35 +61,34 @@ class TestGeometryCollection(TestBase):
"fields": [
{"fieldName": "id", "dataType": "Int64", "isPrimary": True},
{"fieldName": "vector", "dataType": "FloatVector", "elementTypeParams": {"dim": f"{default_dim}"}},
{"fieldName": "geo", "dataType": "Geometry"}
]
{"fieldName": "geo", "dataType": "Geometry"},
],
},
"indexParams": [
{"fieldName": "vector", "indexName": "vector_idx", "metricType": "L2"}
]
"indexParams": [{"fieldName": "vector", "indexName": "vector_idx", "metricType": "L2"}],
}
rsp = self.collection_client.collection_create(payload)
assert rsp['code'] == 0
assert rsp["code"] == 0
# Generate WKT data
nb = default_nb
wkt_data = generate_wkt_by_type(wkt_type, bounds=(0, 100, 0, 100), count=nb)
data = []
for i, wkt in enumerate(wkt_data):
data.append({
"id": i,
"vector": preprocessing.normalize([np.array([random.random() for _ in range(default_dim)])])[0].tolist(),
"geo": wkt
})
data.append(
{
"id": i,
"vector": preprocessing.normalize([np.array([random.random() for _ in range(default_dim)])])[
0
].tolist(),
"geo": wkt,
}
)
# Insert data
insert_payload = {
"collectionName": name,
"data": data
}
insert_payload = {"collectionName": name, "data": data}
rsp = self.vector_client.vector_insert(insert_payload)
assert rsp['code'] == 0
assert rsp['data']['insertCount'] == nb
assert rsp["code"] == 0
assert rsp["data"]["insertCount"] == nb
logger.info(f"Inserted {nb} {wkt_type} geometries")
@pytest.mark.parametrize("index_type", ["RTREE", "AUTOINDEX"])
@@ -115,16 +107,16 @@ class TestGeometryCollection(TestBase):
"fields": [
{"fieldName": "id", "dataType": "Int64", "isPrimary": True},
{"fieldName": "vector", "dataType": "FloatVector", "elementTypeParams": {"dim": f"{default_dim}"}},
{"fieldName": "geo", "dataType": "Geometry"}
]
{"fieldName": "geo", "dataType": "Geometry"},
],
},
"indexParams": [
{"fieldName": "vector", "indexName": "vector_idx", "metricType": "L2"},
{"fieldName": "geo", "indexName": "geo_idx", "indexType": index_type}
]
{"fieldName": "geo", "indexName": "geo_idx", "indexType": index_type},
],
}
rsp = self.collection_client.collection_create(payload)
assert rsp['code'] == 0
assert rsp["code"] == 0
# Insert some geometry data
nb = 50
@@ -132,36 +124,32 @@ class TestGeometryCollection(TestBase):
for i in range(nb):
x = random.uniform(0, 100)
y = random.uniform(0, 100)
data.append({
"id": i,
"vector": preprocessing.normalize([np.array([random.random() for _ in range(default_dim)])])[0].tolist(),
"geo": f"POINT ({x:.2f} {y:.2f})"
})
data.append(
{
"id": i,
"vector": preprocessing.normalize([np.array([random.random() for _ in range(default_dim)])])[
0
].tolist(),
"geo": f"POINT ({x:.2f} {y:.2f})",
}
)
insert_payload = {
"collectionName": name,
"data": data
}
insert_payload = {"collectionName": name, "data": data}
rsp = self.vector_client.vector_insert(insert_payload)
assert rsp['code'] == 0
assert rsp["code"] == 0
# Load collection
self.wait_collection_load_completed(name)
# Verify index
rsp = self.index_client.index_list(name)
assert rsp['code'] == 0
assert rsp["code"] == 0
logger.info(f"Indexes: {rsp}")
@pytest.mark.parametrize("spatial_func", [
"ST_INTERSECTS",
"ST_CONTAINS",
"ST_WITHIN",
"ST_EQUALS",
"ST_TOUCHES",
"ST_OVERLAPS",
"ST_CROSSES"
])
@pytest.mark.parametrize(
"spatial_func",
["ST_INTERSECTS", "ST_CONTAINS", "ST_WITHIN", "ST_EQUALS", "ST_TOUCHES", "ST_OVERLAPS", "ST_CROSSES"],
)
@pytest.mark.parametrize("data_state", ["sealed", "growing", "sealed_and_growing"])
@pytest.mark.parametrize("with_geo_index", [True, False])
@pytest.mark.parametrize("nullable", [True, False])
@@ -188,13 +176,13 @@ class TestGeometryCollection(TestBase):
"fields": [
{"fieldName": "id", "dataType": "Int64", "isPrimary": True},
{"fieldName": "vector", "dataType": "FloatVector", "elementTypeParams": {"dim": f"{default_dim}"}},
geo_field
]
geo_field,
],
},
"indexParams": index_params
"indexParams": index_params,
}
rsp = self.collection_client.collection_create(payload)
assert rsp['code'] == 0
assert rsp["code"] == 0
nb = default_nb
@@ -213,7 +201,9 @@ class TestGeometryCollection(TestBase):
y = 25 + (i // 10) * 5
item = {
"id": start_id + i,
"vector": preprocessing.normalize([np.array([random.random() for _ in range(default_dim)])])[0].tolist(),
"vector": preprocessing.normalize([np.array([random.random() for _ in range(default_dim)])])[
0
].tolist(),
}
if nullable and i % 5 == 0:
item["geo"] = None
@@ -221,7 +211,9 @@ class TestGeometryCollection(TestBase):
item["geo"] = f"POINT ({x:.2f} {y:.2f})"
else:
# Small polygon inside query area
item["geo"] = f"POLYGON (({x:.2f} {y:.2f}, {x + 3:.2f} {y:.2f}, {x + 3:.2f} {y + 3:.2f}, {x:.2f} {y + 3:.2f}, {x:.2f} {y:.2f}))"
item["geo"] = (
f"POLYGON (({x:.2f} {y:.2f}, {x + 3:.2f} {y:.2f}, {x + 3:.2f} {y + 3:.2f}, {x:.2f} {y + 3:.2f}, {x:.2f} {y:.2f}))"
)
data.append(item)
return data
@@ -236,7 +228,9 @@ class TestGeometryCollection(TestBase):
for i in range(count):
item = {
"id": start_id + i,
"vector": preprocessing.normalize([np.array([random.random() for _ in range(default_dim)])])[0].tolist(),
"vector": preprocessing.normalize([np.array([random.random() for _ in range(default_dim)])])[
0
].tolist(),
}
if nullable and i % 5 == 0:
item["geo"] = None
@@ -264,7 +258,9 @@ class TestGeometryCollection(TestBase):
y = 20 + (i // 10) * 6
item = {
"id": start_id + i,
"vector": preprocessing.normalize([np.array([random.random() for _ in range(default_dim)])])[0].tolist(),
"vector": preprocessing.normalize([np.array([random.random() for _ in range(default_dim)])])[
0
].tolist(),
}
if nullable and i % 5 == 0:
item["geo"] = None
@@ -283,7 +279,9 @@ class TestGeometryCollection(TestBase):
for i in range(count):
item = {
"id": start_id + i,
"vector": preprocessing.normalize([np.array([random.random() for _ in range(default_dim)])])[0].tolist(),
"vector": preprocessing.normalize([np.array([random.random() for _ in range(default_dim)])])[
0
].tolist(),
}
if nullable and i % 5 == 0:
item["geo"] = None
@@ -307,7 +305,9 @@ class TestGeometryCollection(TestBase):
for i in range(count):
item = {
"id": start_id + i,
"vector": preprocessing.normalize([np.array([random.random() for _ in range(default_dim)])])[0].tolist(),
"vector": preprocessing.normalize([np.array([random.random() for _ in range(default_dim)])])[
0
].tolist(),
}
if nullable and i % 5 == 0:
item["geo"] = None
@@ -336,7 +336,9 @@ class TestGeometryCollection(TestBase):
for i in range(count):
item = {
"id": start_id + i,
"vector": preprocessing.normalize([np.array([random.random() for _ in range(default_dim)])])[0].tolist(),
"vector": preprocessing.normalize([np.array([random.random() for _ in range(default_dim)])])[
0
].tolist(),
}
if nullable and i % 5 == 0:
item["geo"] = None
@@ -346,10 +348,14 @@ class TestGeometryCollection(TestBase):
offset = (i % 4) * 5
if i % 2 == 0:
# Overlapping from right side
item["geo"] = f"POLYGON (({50 + offset} 45, {70 + offset} 45, {70 + offset} 55, {50 + offset} 55, {50 + offset} 45))"
item["geo"] = (
f"POLYGON (({50 + offset} 45, {70 + offset} 45, {70 + offset} 55, {50 + offset} 55, {50 + offset} 45))"
)
else:
# Overlapping from bottom
item["geo"] = f"POLYGON ((45 {50 + offset}, 55 {50 + offset}, 55 {70 + offset}, 45 {70 + offset}, 45 {50 + offset}))"
item["geo"] = (
f"POLYGON ((45 {50 + offset}, 55 {50 + offset}, 55 {70 + offset}, 45 {70 + offset}, 45 {50 + offset}))"
)
data.append(item)
return data
@@ -363,7 +369,9 @@ class TestGeometryCollection(TestBase):
for i in range(count):
item = {
"id": start_id + i,
"vector": preprocessing.normalize([np.array([random.random() for _ in range(default_dim)])])[0].tolist(),
"vector": preprocessing.normalize([np.array([random.random() for _ in range(default_dim)])])[
0
].tolist(),
}
if nullable and i % 5 == 0:
item["geo"] = None
@@ -384,7 +392,9 @@ class TestGeometryCollection(TestBase):
y = 30 + (i // 10) * 4
item = {
"id": start_id + i,
"vector": preprocessing.normalize([np.array([random.random() for _ in range(default_dim)])])[0].tolist(),
"vector": preprocessing.normalize([np.array([random.random() for _ in range(default_dim)])])[
0
].tolist(),
}
if nullable and i % 5 == 0:
item["geo"] = None
@@ -398,7 +408,7 @@ class TestGeometryCollection(TestBase):
data = generate_geo_data(0, nb)
insert_payload = {"collectionName": name, "data": data}
rsp = self.vector_client.vector_insert(insert_payload)
assert rsp['code'] == 0
assert rsp["code"] == 0
rsp = self.collection_client.flush(name)
self.wait_collection_load_completed(name)
@@ -407,33 +417,30 @@ class TestGeometryCollection(TestBase):
data = generate_geo_data(0, nb)
insert_payload = {"collectionName": name, "data": data}
rsp = self.vector_client.vector_insert(insert_payload)
assert rsp['code'] == 0
assert rsp["code"] == 0
else: # sealed_and_growing
sealed_data = generate_geo_data(0, nb // 2)
insert_payload = {"collectionName": name, "data": sealed_data}
rsp = self.vector_client.vector_insert(insert_payload)
assert rsp['code'] == 0
assert rsp["code"] == 0
rsp = self.collection_client.flush(name)
self.wait_collection_load_completed(name)
growing_data = generate_geo_data(nb // 2, nb // 2)
insert_payload = {"collectionName": name, "data": growing_data}
rsp = self.vector_client.vector_insert(insert_payload)
assert rsp['code'] == 0
assert rsp["code"] == 0
filter_expr = f"{spatial_func}(geo, '{query_geom}')"
# 1. Query with spatial filter
query_payload = {
"collectionName": name,
"filter": filter_expr,
"outputFields": ["id", "geo"],
"limit": 100
}
query_payload = {"collectionName": name, "filter": filter_expr, "outputFields": ["id", "geo"], "limit": 100}
rsp = self.vector_client.vector_query(query_payload)
assert rsp['code'] == 0
query_count = len(rsp.get('data', []))
logger.info(f"{spatial_func} ({data_state}, geo_index={with_geo_index}, nullable={nullable}) query returned {query_count} results")
assert rsp["code"] == 0
query_count = len(rsp.get("data", []))
logger.info(
f"{spatial_func} ({data_state}, geo_index={with_geo_index}, nullable={nullable}) query returned {query_count} results"
)
# Verify we got results (except for edge cases)
if not nullable or spatial_func not in ["ST_EQUALS"]:
assert query_count > 0, f"{spatial_func} query should return results"
@@ -446,12 +453,14 @@ class TestGeometryCollection(TestBase):
"annsField": "vector",
"filter": filter_expr,
"limit": 10,
"outputFields": ["id", "geo"]
"outputFields": ["id", "geo"],
}
rsp = self.vector_client.vector_search(search_payload)
assert rsp['code'] == 0
search_count = len(rsp.get('data', []))
logger.info(f"{spatial_func} ({data_state}, geo_index={with_geo_index}, nullable={nullable}) search returned {search_count} results")
assert rsp["code"] == 0
search_count = len(rsp.get("data", []))
logger.info(
f"{spatial_func} ({data_state}, geo_index={with_geo_index}, nullable={nullable}) search returned {search_count} results"
)
def test_upsert_geometry_data(self):
"""
@@ -468,16 +477,16 @@ class TestGeometryCollection(TestBase):
"fields": [
{"fieldName": "id", "dataType": "Int64", "isPrimary": True},
{"fieldName": "vector", "dataType": "FloatVector", "elementTypeParams": {"dim": f"{default_dim}"}},
{"fieldName": "geo", "dataType": "Geometry"}
]
{"fieldName": "geo", "dataType": "Geometry"},
],
},
"indexParams": [
{"fieldName": "vector", "indexName": "vector_idx", "metricType": "L2"},
{"fieldName": "geo", "indexName": "geo_idx", "indexType": "RTREE"}
]
{"fieldName": "geo", "indexName": "geo_idx", "indexType": "RTREE"},
],
}
rsp = self.collection_client.collection_create(payload)
assert rsp['code'] == 0
assert rsp["code"] == 0
nb = default_nb
@@ -486,25 +495,29 @@ class TestGeometryCollection(TestBase):
for i in range(count):
x = random.uniform(10, 90)
y = random.uniform(10, 90)
data.append({
"id": start_id + i,
"vector": preprocessing.normalize([np.array([random.random() for _ in range(default_dim)])])[0].tolist(),
"geo": f"POINT ({x:.2f} {y:.2f})"
})
data.append(
{
"id": start_id + i,
"vector": preprocessing.normalize([np.array([random.random() for _ in range(default_dim)])])[
0
].tolist(),
"geo": f"POINT ({x:.2f} {y:.2f})",
}
)
return data
# Insert initial data
data = generate_geo_data(0, nb)
insert_payload = {"collectionName": name, "data": data}
rsp = self.vector_client.vector_insert(insert_payload)
assert rsp['code'] == 0
assert rsp["code"] == 0
self.wait_collection_load_completed(name)
# Upsert data
upsert_data = generate_geo_data(0, nb // 2)
upsert_payload = {"collectionName": name, "data": upsert_data}
rsp = self.vector_client.vector_upsert(upsert_payload)
assert rsp['code'] == 0
assert rsp["code"] == 0
logger.info("Upsert geometry data completed successfully")
def test_delete_geometry_data(self):
@@ -522,16 +535,16 @@ class TestGeometryCollection(TestBase):
"fields": [
{"fieldName": "id", "dataType": "Int64", "isPrimary": True},
{"fieldName": "vector", "dataType": "FloatVector", "elementTypeParams": {"dim": f"{default_dim}"}},
{"fieldName": "geo", "dataType": "Geometry"}
]
{"fieldName": "geo", "dataType": "Geometry"},
],
},
"indexParams": [
{"fieldName": "vector", "indexName": "vector_idx", "metricType": "L2"},
{"fieldName": "geo", "indexName": "geo_idx", "indexType": "RTREE"}
]
{"fieldName": "geo", "indexName": "geo_idx", "indexType": "RTREE"},
],
}
rsp = self.collection_client.collection_create(payload)
assert rsp['code'] == 0
assert rsp["code"] == 0
nb = default_nb
@@ -540,35 +553,34 @@ class TestGeometryCollection(TestBase):
for i in range(count):
x = random.uniform(10, 90)
y = random.uniform(10, 90)
data.append({
"id": start_id + i,
"vector": preprocessing.normalize([np.array([random.random() for _ in range(default_dim)])])[0].tolist(),
"geo": f"POINT ({x:.2f} {y:.2f})"
})
data.append(
{
"id": start_id + i,
"vector": preprocessing.normalize([np.array([random.random() for _ in range(default_dim)])])[
0
].tolist(),
"geo": f"POINT ({x:.2f} {y:.2f})",
}
)
return data
# Insert data
data = generate_geo_data(0, nb)
insert_payload = {"collectionName": name, "data": data}
rsp = self.vector_client.vector_insert(insert_payload)
assert rsp['code'] == 0
assert rsp["code"] == 0
self.wait_collection_load_completed(name)
# Delete data
delete_ids = list(range(0, nb // 2))
delete_payload = {"collectionName": name, "filter": f"id in {delete_ids}"}
rsp = self.vector_client.vector_delete(delete_payload)
assert rsp['code'] == 0
assert rsp["code"] == 0
# Verify deletion by querying
query_payload = {
"collectionName": name,
"filter": "id >= 0",
"outputFields": ["id", "geo"],
"limit": 200
}
query_payload = {"collectionName": name, "filter": "id >= 0", "outputFields": ["id", "geo"], "limit": 200}
rsp = self.vector_client.vector_query(query_payload)
assert rsp['code'] == 0
assert rsp["code"] == 0
logger.info(f"Delete geometry data completed, remaining: {len(rsp.get('data', []))} records")
def test_geometry_default_value(self):
@@ -587,23 +599,25 @@ class TestGeometryCollection(TestBase):
"fields": [
{"fieldName": "id", "dataType": "Int64", "isPrimary": True},
{"fieldName": "vector", "dataType": "FloatVector", "elementTypeParams": {"dim": f"{default_dim}"}},
{"fieldName": "geo", "dataType": "Geometry", "defaultValue": default_geo}
]
{"fieldName": "geo", "dataType": "Geometry", "defaultValue": default_geo},
],
},
"indexParams": [
{"fieldName": "vector", "indexName": "vector_idx", "metricType": "L2"},
{"fieldName": "geo", "indexName": "geo_idx", "indexType": "RTREE"}
]
{"fieldName": "geo", "indexName": "geo_idx", "indexType": "RTREE"},
],
}
rsp = self.collection_client.collection_create(payload)
assert rsp['code'] == 0
assert rsp["code"] == 0
nb = default_nb
data = []
for i in range(nb):
item = {
"id": i,
"vector": preprocessing.normalize([np.array([random.random() for _ in range(default_dim)])])[0].tolist(),
"vector": preprocessing.normalize([np.array([random.random() for _ in range(default_dim)])])[
0
].tolist(),
}
# 30% use default value (omit geo field)
if i % 3 != 0:
@@ -615,7 +629,7 @@ class TestGeometryCollection(TestBase):
insert_payload = {"collectionName": name, "data": data}
rsp = self.vector_client.vector_insert(insert_payload)
assert rsp['code'] == 0
assert rsp["code"] == 0
self.wait_collection_load_completed(name)
# Query for records with default geometry value
@@ -623,23 +637,18 @@ class TestGeometryCollection(TestBase):
"collectionName": name,
"filter": f"ST_EQUALS(geo, '{default_geo}')",
"outputFields": ["id", "geo"],
"limit": 100
"limit": 100,
}
rsp = self.vector_client.vector_query(query_payload)
assert rsp['code'] == 0
default_count = len(rsp.get('data', []))
assert rsp["code"] == 0
default_count = len(rsp.get("data", []))
logger.info(f"Default geometry: found {default_count} records with default value")
# Query all records
query_payload = {
"collectionName": name,
"filter": "id >= 0",
"outputFields": ["id", "geo"],
"limit": 200
}
query_payload = {"collectionName": name, "filter": "id >= 0", "outputFields": ["id", "geo"], "limit": 200}
rsp = self.vector_client.vector_query(query_payload)
assert rsp['code'] == 0
total_count = len(rsp.get('data', []))
assert rsp["code"] == 0
total_count = len(rsp.get("data", []))
logger.info(f"Default geometry: total {total_count} records")
# Spatial query with default value area
@@ -647,17 +656,20 @@ class TestGeometryCollection(TestBase):
"collectionName": name,
"filter": "ST_WITHIN(geo, 'POLYGON ((-5 -5, 5 -5, 5 5, -5 5, -5 -5))')",
"outputFields": ["id", "geo"],
"limit": 100
"limit": 100,
}
rsp = self.vector_client.vector_query(query_payload)
assert rsp['code'] == 0
assert rsp["code"] == 0
logger.info(f"Default geometry: spatial query near origin returned {len(rsp.get('data', []))} results")
@pytest.mark.parametrize("spatial_func", [
"ST_INTERSECTS",
"ST_CONTAINS",
"ST_WITHIN",
])
@pytest.mark.parametrize(
"spatial_func",
[
"ST_INTERSECTS",
"ST_CONTAINS",
"ST_WITHIN",
],
)
def test_spatial_query_empty_result(self, spatial_func):
"""
target: test spatial query returns empty result when no data matches
@@ -673,16 +685,16 @@ class TestGeometryCollection(TestBase):
"fields": [
{"fieldName": "id", "dataType": "Int64", "isPrimary": True},
{"fieldName": "vector", "dataType": "FloatVector", "elementTypeParams": {"dim": f"{default_dim}"}},
{"fieldName": "geo", "dataType": "Geometry"}
]
{"fieldName": "geo", "dataType": "Geometry"},
],
},
"indexParams": [
{"fieldName": "vector", "indexName": "vector_idx", "metricType": "L2"},
{"fieldName": "geo", "indexName": "geo_idx", "indexType": "RTREE"}
]
{"fieldName": "geo", "indexName": "geo_idx", "indexType": "RTREE"},
],
}
rsp = self.collection_client.collection_create(payload)
assert rsp['code'] == 0
assert rsp["code"] == 0
# Insert data in region (0-50, 0-50)
nb = 50
@@ -690,15 +702,19 @@ class TestGeometryCollection(TestBase):
for i in range(nb):
x = 10 + (i % 10) * 4
y = 10 + (i // 10) * 4
data.append({
"id": i,
"vector": preprocessing.normalize([np.array([random.random() for _ in range(default_dim)])])[0].tolist(),
"geo": f"POINT ({x:.2f} {y:.2f})"
})
data.append(
{
"id": i,
"vector": preprocessing.normalize([np.array([random.random() for _ in range(default_dim)])])[
0
].tolist(),
"geo": f"POINT ({x:.2f} {y:.2f})",
}
)
insert_payload = {"collectionName": name, "data": data}
rsp = self.vector_client.vector_insert(insert_payload)
assert rsp['code'] == 0
assert rsp["code"] == 0
self.wait_collection_load_completed(name)
# Query with geometry far away from all data (region 200-300, 200-300)
@@ -712,14 +728,9 @@ class TestGeometryCollection(TestBase):
query_geom = "POLYGON ((200 200, 300 200, 300 300, 200 300, 200 200))"
filter_expr = f"{spatial_func}(geo, '{query_geom}')"
query_payload = {
"collectionName": name,
"filter": filter_expr,
"outputFields": ["id", "geo"],
"limit": 100
}
query_payload = {"collectionName": name, "filter": filter_expr, "outputFields": ["id", "geo"], "limit": 100}
rsp = self.vector_client.vector_query(query_payload)
assert rsp['code'] == 0
result_count = len(rsp.get('data', []))
assert rsp["code"] == 0
result_count = len(rsp.get("data", []))
logger.info(f"{spatial_func} empty result test: query returned {result_count} results")
assert result_count == 0, f"{spatial_func} query should return empty result when no data matches"
File diff suppressed because it is too large Load Diff
File diff suppressed because it is too large Load Diff
@@ -8,6 +8,7 @@ from base.testbase import TestBase
from faker import Faker
from pymilvus import Collection
from sklearn import preprocessing
from utils.constant import CaseLabel
from utils.util_log import test_log as logger
from utils.utils import (
en_vocabularies_distribution,
@@ -33,7 +34,7 @@ index_param_map = {
}
@pytest.mark.L0
@pytest.mark.tags(CaseLabel.L0)
class TestCreateIndex(TestBase):
@pytest.mark.parametrize("metric_type", ["L2", "COSINE", "IP"])
@pytest.mark.parametrize("index_type", ["AUTOINDEX", "IVF_SQ8", "HNSW"])
@@ -477,7 +478,7 @@ class TestCreateIndex(TestBase):
assert info["index_param"]["index_type"] == index_type
@pytest.mark.L0
@pytest.mark.tags(CaseLabel.L0)
class TestIndexProperties(TestBase):
"""Test index properties operations"""
@@ -621,7 +622,7 @@ class TestIndexProperties(TestBase):
assert rsp["code"] == 1100
@pytest.mark.L1
@pytest.mark.tags(CaseLabel.L1)
class TestCreateIndexNegative(TestBase):
@pytest.mark.parametrize("index_type", ["BIN_FLAT", "BIN_IVF_FLAT"])
@pytest.mark.parametrize("metric_type", ["L2", "IP", "COSINE"])
File diff suppressed because it is too large Load Diff
File diff suppressed because it is too large Load Diff
@@ -1,17 +1,16 @@
import random
from sklearn import preprocessing
import numpy as np
from utils.utils import gen_collection_name
import pytest
from base.testbase import TestBase
from pymilvus import (
Collection
)
from pymilvus import Collection
from sklearn import preprocessing
from utils.constant import CaseLabel
from utils.utils import gen_collection_name
@pytest.mark.L0
@pytest.mark.tags(CaseLabel.L0)
class TestPartitionE2E(TestBase):
def test_partition_e2e(self):
"""
target: test create collection
@@ -29,21 +28,22 @@ class TestPartitionE2E(TestBase):
{"fieldName": "book_id", "dataType": "Int64", "isPrimary": True, "elementTypeParams": {}},
{"fieldName": "word_count", "dataType": "Int64", "elementTypeParams": {}},
{"fieldName": "book_describe", "dataType": "VarChar", "elementTypeParams": {"max_length": "256"}},
{"fieldName": "book_intro", "dataType": "FloatVector", "elementTypeParams": {"dim": f"{dim}"}}
{"fieldName": "book_intro", "dataType": "FloatVector", "elementTypeParams": {"dim": f"{dim}"}},
]
},
"indexParams": [
{"fieldName": "book_intro", "indexName": "book_intro_vector", "metricType": f"{metric_type}"}]
{"fieldName": "book_intro", "indexName": "book_intro_vector", "metricType": f"{metric_type}"}
],
}
rsp = self.collection_client.collection_create(payload)
assert rsp['code'] == 0
assert rsp["code"] == 0
rsp = client.collection_list()
all_collections = rsp['data']
all_collections = rsp["data"]
assert name in all_collections
# describe collection
rsp = client.collection_describe(name)
assert rsp['code'] == 0
assert rsp['data']['collectionName'] == name
assert rsp["code"] == 0
assert rsp["data"]["collectionName"] == name
# insert data to default partition
data = []
for j in range(3000):
@@ -51,7 +51,7 @@ class TestPartitionE2E(TestBase):
"book_id": j,
"word_count": j,
"book_describe": f"book_{j}",
"book_intro": preprocessing.normalize([np.array([random.random() for i in range(dim)])])[0].tolist()
"book_intro": preprocessing.normalize([np.array([random.random() for i in range(dim)])])[0].tolist(),
}
data.append(tmp)
payload = {
@@ -59,11 +59,11 @@ class TestPartitionE2E(TestBase):
"data": data,
}
rsp = self.vector_client.vector_insert(payload)
assert rsp['code'] == 0
assert rsp["code"] == 0
# create partition
partition_name = "test_partition"
rsp = self.partition_client.partition_create(collection_name=name, partition_name=partition_name)
assert rsp['code'] == 0
assert rsp["code"] == 0
# insert data to partition
data = []
for j in range(3000, 6000):
@@ -71,7 +71,7 @@ class TestPartitionE2E(TestBase):
"book_id": j,
"word_count": j,
"book_describe": f"book_{j}",
"book_intro": preprocessing.normalize([np.array([random.random() for i in range(dim)])])[0].tolist()
"book_intro": preprocessing.normalize([np.array([random.random() for i in range(dim)])])[0].tolist(),
}
data.append(tmp)
payload = {
@@ -80,45 +80,45 @@ class TestPartitionE2E(TestBase):
"data": data,
}
rsp = self.vector_client.vector_insert(payload)
assert rsp['code'] == 0
assert rsp["code"] == 0
# create partition again
rsp = self.partition_client.partition_create(collection_name=name, partition_name=partition_name)
# list partitions
rsp = self.partition_client.partition_list(collection_name=name)
assert rsp['code'] == 0
assert partition_name in rsp['data']
assert rsp["code"] == 0
assert partition_name in rsp["data"]
# has partition
rsp = self.partition_client.partition_has(collection_name=name, partition_name=partition_name)
assert rsp['code'] == 0
assert rsp['data']["has"] is True
assert rsp["code"] == 0
assert rsp["data"]["has"] is True
# flush and get partition statistics
c = Collection(name=name)
c.flush()
rsp = self.partition_client.partition_stats(collection_name=name, partition_name=partition_name)
assert rsp['code'] == 0
assert rsp['data']['rowCount'] == 3000
assert rsp["code"] == 0
assert rsp["data"]["rowCount"] == 3000
# release partition
rsp = self.partition_client.partition_release(collection_name=name, partition_names=[partition_name])
assert rsp['code'] == 0
assert rsp["code"] == 0
# release partition again
rsp = self.partition_client.partition_release(collection_name=name, partition_names=[partition_name])
assert rsp['code'] == 0
assert rsp["code"] == 0
# load partition
rsp = self.partition_client.partition_load(collection_name=name, partition_names=[partition_name])
assert rsp['code'] == 0
assert rsp["code"] == 0
# load partition again
rsp = self.partition_client.partition_load(collection_name=name, partition_names=[partition_name])
assert rsp['code'] == 0
assert rsp["code"] == 0
# drop partition when it is loaded
rsp = self.partition_client.partition_drop(collection_name=name, partition_name=partition_name)
assert rsp['code'] == 1100
assert rsp["code"] == 1100
# drop partition after release
rsp = self.partition_client.partition_release(collection_name=name, partition_names=[partition_name])
rsp = self.partition_client.partition_drop(collection_name=name, partition_name=partition_name)
assert rsp['code'] == 0
assert rsp["code"] == 0
# has partition
rsp = self.partition_client.partition_has(collection_name=name, partition_name=partition_name)
assert rsp['code'] == 0
assert rsp['data']["has"] is False
assert rsp["code"] == 0
assert rsp["data"]["has"] is False
@@ -1,54 +1,50 @@
import random
import time
from utils.utils import gen_collection_name
from utils.util_log import test_log as logger
from utils.constant import default_nb
import pytest
from base.testbase import TestBase
from pymilvus import (
FieldSchema, CollectionSchema, DataType,
Collection
)
from pymilvus import Collection, CollectionSchema, DataType, FieldSchema
from utils.constant import CaseLabel, default_nb
from utils.util_log import test_log as logger
from utils.utils import gen_collection_name
@pytest.mark.L0
@pytest.mark.tags(CaseLabel.L0)
class TestRestfulSdkCompatibility(TestBase):
@pytest.mark.parametrize("dim", [128, 256])
@pytest.mark.parametrize("enable_dynamic", [True, False])
@pytest.mark.parametrize("num_shards", [1, 2])
def test_collection_created_by_sdk_describe_by_restful(self, dim, enable_dynamic, num_shards):
"""
"""
""" """
# 1. create collection by sdk
name = gen_collection_name()
default_fields = [
FieldSchema(name="int64", dtype=DataType.INT64, is_primary=True),
FieldSchema(name="float", dtype=DataType.FLOAT),
FieldSchema(name="varchar", dtype=DataType.VARCHAR, max_length=65535),
FieldSchema(name="float_vector", dtype=DataType.FLOAT_VECTOR, dim=dim)
FieldSchema(name="float_vector", dtype=DataType.FLOAT_VECTOR, dim=dim),
]
default_schema = CollectionSchema(fields=default_fields, description="test collection",
enable_dynamic_field=enable_dynamic)
default_schema = CollectionSchema(
fields=default_fields, description="test collection", enable_dynamic_field=enable_dynamic
)
collection = Collection(name=name, schema=default_schema, num_shards=num_shards)
logger.info(collection.schema)
# 2. use restful to get collection info
client = self.collection_client
rsp = client.collection_list()
all_collections = rsp['data']
all_collections = rsp["data"]
assert name in all_collections
rsp = client.collection_describe(name)
assert rsp['code'] == 0
assert rsp['data']['collectionName'] == name
assert rsp['data']['enableDynamicField'] == enable_dynamic
assert rsp['data']['load'] == "LoadStateNotLoad"
assert rsp['data']['shardsNum'] == num_shards
assert rsp["code"] == 0
assert rsp["data"]["collectionName"] == name
assert rsp["data"]["enableDynamicField"] == enable_dynamic
assert rsp["data"]["load"] == "LoadStateNotLoad"
assert rsp["data"]["shardsNum"] == num_shards
@pytest.mark.parametrize("metric_type", ["L2", "IP", "COSINE"])
@pytest.mark.parametrize("dim", [128])
def test_collection_created_by_restful_describe_by_sdk(self, dim, metric_type):
"""
"""
""" """
name = gen_collection_name()
dim = 128
client = self.collection_client
@@ -58,7 +54,7 @@ class TestRestfulSdkCompatibility(TestBase):
"metricType": metric_type,
}
rsp = client.collection_create(payload)
assert rsp['code'] == 0
assert rsp["code"] == 0
collection = Collection(name=name)
logger.info(collection.schema)
field_names = [field.name for field in collection.schema.fields]
@@ -68,18 +64,18 @@ class TestRestfulSdkCompatibility(TestBase):
@pytest.mark.parametrize("metric_type", ["L2", "IP"])
def test_collection_created_index_by_sdk_describe_by_restful(self, metric_type):
"""
"""
""" """
# 1. create collection by sdk
name = gen_collection_name()
default_fields = [
FieldSchema(name="int64", dtype=DataType.INT64, is_primary=True),
FieldSchema(name="float", dtype=DataType.FLOAT),
FieldSchema(name="varchar", dtype=DataType.VARCHAR, max_length=65535),
FieldSchema(name="float_vector", dtype=DataType.FLOAT_VECTOR, dim=128)
FieldSchema(name="float_vector", dtype=DataType.FLOAT_VECTOR, dim=128),
]
default_schema = CollectionSchema(fields=default_fields, description="test collection",
enable_dynamic_field=True)
default_schema = CollectionSchema(
fields=default_fields, description="test collection", enable_dynamic_field=True
)
collection = Collection(name=name, schema=default_schema)
# create index by sdk
index_param = {"metric_type": metric_type, "index_type": "IVF_FLAT", "params": {"nlist": 128}}
@@ -87,27 +83,27 @@ class TestRestfulSdkCompatibility(TestBase):
# 2. use restful to get collection info
client = self.collection_client
rsp = client.collection_list()
all_collections = rsp['data']
all_collections = rsp["data"]
assert name in all_collections
rsp = client.collection_describe(name)
assert rsp['code'] == 0
assert rsp['data']['collectionName'] == name
assert len(rsp['data']['indexes']) == 1 and rsp['data']['indexes'][0]['metricType'] == metric_type
assert rsp["code"] == 0
assert rsp["data"]["collectionName"] == name
assert len(rsp["data"]["indexes"]) == 1 and rsp["data"]["indexes"][0]["metricType"] == metric_type
@pytest.mark.parametrize("metric_type", ["L2", "IP"])
def test_collection_load_by_sdk_describe_by_restful(self, metric_type):
"""
"""
""" """
# 1. create collection by sdk
name = gen_collection_name()
default_fields = [
FieldSchema(name="int64", dtype=DataType.INT64, is_primary=True),
FieldSchema(name="float", dtype=DataType.FLOAT),
FieldSchema(name="varchar", dtype=DataType.VARCHAR, max_length=65535),
FieldSchema(name="float_vector", dtype=DataType.FLOAT_VECTOR, dim=128)
FieldSchema(name="float_vector", dtype=DataType.FLOAT_VECTOR, dim=128),
]
default_schema = CollectionSchema(fields=default_fields, description="test collection",
enable_dynamic_field=True)
default_schema = CollectionSchema(
fields=default_fields, description="test collection", enable_dynamic_field=True
)
collection = Collection(name=name, schema=default_schema)
# create index by sdk
index_param = {"metric_type": metric_type, "index_type": "IVF_FLAT", "params": {"nlist": 128}}
@@ -116,14 +112,13 @@ class TestRestfulSdkCompatibility(TestBase):
# 2. use restful to get collection info
client = self.collection_client
rsp = client.collection_list()
all_collections = rsp['data']
all_collections = rsp["data"]
assert name in all_collections
rsp = client.collection_describe(name)
assert rsp['data']['load'] == "LoadStateLoaded"
assert rsp["data"]["load"] == "LoadStateLoaded"
def test_collection_create_by_sdk_insert_vector_by_restful(self):
"""
"""
""" """
# 1. create collection by sdk
dim = 128
nb = default_nb
@@ -134,11 +129,18 @@ class TestRestfulSdkCompatibility(TestBase):
FieldSchema(name="varchar", dtype=DataType.VARCHAR, max_length=65535),
FieldSchema(name="json", dtype=DataType.JSON),
FieldSchema(name="int_array", dtype=DataType.ARRAY, element_type=DataType.INT64, max_capacity=1024),
FieldSchema(name="varchar_array", dtype=DataType.ARRAY, element_type=DataType.VARCHAR, max_capacity=1024, max_length=65535),
FieldSchema(name="float_vector", dtype=DataType.FLOAT_VECTOR, dim=128)
FieldSchema(
name="varchar_array",
dtype=DataType.ARRAY,
element_type=DataType.VARCHAR,
max_capacity=1024,
max_length=65535,
),
FieldSchema(name="float_vector", dtype=DataType.FLOAT_VECTOR, dim=128),
]
default_schema = CollectionSchema(fields=default_fields, description="test collection",
enable_dynamic_field=True)
default_schema = CollectionSchema(
fields=default_fields, description="test collection", enable_dynamic_field=True
)
collection = Collection(name=name, schema=default_schema)
# create index by sdk
index_param = {"metric_type": "L2", "index_type": "IVF_FLAT", "params": {"nlist": 128}}
@@ -146,13 +148,16 @@ class TestRestfulSdkCompatibility(TestBase):
collection.load()
# insert data by restful
data = [
{"int64": i,
"float": i,
"varchar": str(i),
"json": {f"key_{i}": f"value_{i}"},
"int_array": [random.randint(0, 100) for _ in range(10)],
"varchar_array": [str(i) for _ in range(10)],
"float_vector": [random.random() for _ in range(dim)], "age": i}
{
"int64": i,
"float": i,
"varchar": str(i),
"json": {f"key_{i}": f"value_{i}"},
"int_array": [random.randint(0, 100) for _ in range(10)],
"varchar_array": [str(i) for _ in range(10)],
"float_vector": [random.random() for _ in range(dim)],
"age": i,
}
for i in range(nb)
]
client = self.vector_client
@@ -161,13 +166,12 @@ class TestRestfulSdkCompatibility(TestBase):
"data": data,
}
rsp = client.vector_insert(payload)
assert rsp['code'] == 0
assert rsp['data']['insertCount'] == nb
assert len(rsp['data']["insertIds"]) == nb
assert rsp["code"] == 0
assert rsp["data"]["insertCount"] == nb
assert len(rsp["data"]["insertIds"]) == nb
def test_collection_create_by_sdk_search_vector_by_restful(self):
"""
"""
""" """
dim = 128
nb = default_nb
name = gen_collection_name()
@@ -175,10 +179,11 @@ class TestRestfulSdkCompatibility(TestBase):
FieldSchema(name="int64", dtype=DataType.INT64, is_primary=True),
FieldSchema(name="float", dtype=DataType.FLOAT),
FieldSchema(name="varchar", dtype=DataType.VARCHAR, max_length=65535),
FieldSchema(name="float_vector", dtype=DataType.FLOAT_VECTOR, dim=128)
FieldSchema(name="float_vector", dtype=DataType.FLOAT_VECTOR, dim=128),
]
default_schema = CollectionSchema(fields=default_fields, description="test collection",
enable_dynamic_field=True)
default_schema = CollectionSchema(
fields=default_fields, description="test collection", enable_dynamic_field=True
)
# init collection by sdk
collection = Collection(name=name, schema=default_schema)
index_param = {"metric_type": "L2", "index_type": "IVF_FLAT", "params": {"nlist": 128}}
@@ -190,19 +195,14 @@ class TestRestfulSdkCompatibility(TestBase):
]
collection.insert(data)
client = self.vector_client
payload = {
"collectionName": name,
"data": [[random.random() for _ in range(dim)]],
"limit": 10
}
payload = {"collectionName": name, "data": [[random.random() for _ in range(dim)]], "limit": 10}
# search data by restful
rsp = client.vector_search(payload)
assert rsp['code'] == 0
assert len(rsp['data']) == 10
assert rsp["code"] == 0
assert len(rsp["data"]) == 10
def test_collection_create_by_sdk_query_vector_by_restful(self):
"""
"""
""" """
dim = 128
nb = default_nb
name = gen_collection_name()
@@ -210,10 +210,11 @@ class TestRestfulSdkCompatibility(TestBase):
FieldSchema(name="int64", dtype=DataType.INT64, is_primary=True),
FieldSchema(name="float", dtype=DataType.FLOAT),
FieldSchema(name="varchar", dtype=DataType.VARCHAR, max_length=65535),
FieldSchema(name="float_vector", dtype=DataType.FLOAT_VECTOR, dim=128)
FieldSchema(name="float_vector", dtype=DataType.FLOAT_VECTOR, dim=128),
]
default_schema = CollectionSchema(fields=default_fields, description="test collection",
enable_dynamic_field=True)
default_schema = CollectionSchema(
fields=default_fields, description="test collection", enable_dynamic_field=True
)
# init collection by sdk
collection = Collection(name=name, schema=default_schema)
index_param = {"metric_type": "L2", "index_type": "IVF_FLAT", "params": {"nlist": 128}}
@@ -231,12 +232,11 @@ class TestRestfulSdkCompatibility(TestBase):
}
# query data by restful
rsp = client.vector_query(payload)
assert rsp['code'] == 0
assert len(rsp['data']) == 10
assert rsp["code"] == 0
assert len(rsp["data"]) == 10
def test_collection_create_by_restful_search_vector_by_sdk(self):
"""
"""
""" """
name = gen_collection_name()
dim = 128
# insert data by restful
@@ -251,8 +251,7 @@ class TestRestfulSdkCompatibility(TestBase):
assert len(res[0]) == 10
def test_collection_create_by_restful_query_vector_by_sdk(self):
"""
"""
""" """
name = gen_collection_name()
dim = 128
# insert data by restful
@@ -260,14 +259,13 @@ class TestRestfulSdkCompatibility(TestBase):
time.sleep(5)
# query data by sdk
collection = Collection(name=name)
res = collection.query(expr=f"uid in [0, 1, 2, 3, 4, 5, 6, 7, 8, 9]", output_fields=["*"])
res = collection.query(expr="uid in [0, 1, 2, 3, 4, 5, 6, 7, 8, 9]", output_fields=["*"])
for item in res:
uid = item["uid"]
assert uid in [0, 1, 2, 3, 4, 5, 6, 7, 8, 9]
def test_collection_create_by_restful_delete_vector_by_sdk(self):
"""
"""
""" """
name = gen_collection_name()
dim = 128
# insert data by restful
@@ -275,20 +273,18 @@ class TestRestfulSdkCompatibility(TestBase):
time.sleep(5)
# query data by sdk
collection = Collection(name=name)
res = collection.query(expr=f"uid in [0, 1, 2, 3, 4, 5, 6, 7, 8, 9]", output_fields=["*"])
res = collection.query(expr="uid in [0, 1, 2, 3, 4, 5, 6, 7, 8, 9]", output_fields=["*"])
pk_id_list = []
for item in res:
uid = item["uid"]
pk_id_list.append(item["id"])
expr = f"id in {pk_id_list}"
collection.delete(expr)
time.sleep(5)
res = collection.query(expr=f"uid in [0, 1, 2, 3, 4, 5, 6, 7, 8, 9]", output_fields=["*"])
res = collection.query(expr="uid in [0, 1, 2, 3, 4, 5, 6, 7, 8, 9]", output_fields=["*"])
assert len(res) == 0
def test_collection_create_by_sdk_delete_vector_by_restful(self):
"""
"""
""" """
dim = 128
nb = default_nb
name = gen_collection_name()
@@ -296,10 +292,11 @@ class TestRestfulSdkCompatibility(TestBase):
FieldSchema(name="int64", dtype=DataType.INT64, is_primary=True),
FieldSchema(name="float", dtype=DataType.FLOAT),
FieldSchema(name="varchar", dtype=DataType.VARCHAR, max_length=65535),
FieldSchema(name="float_vector", dtype=DataType.FLOAT_VECTOR, dim=128)
FieldSchema(name="float_vector", dtype=DataType.FLOAT_VECTOR, dim=128),
]
default_schema = CollectionSchema(fields=default_fields, description="test collection",
enable_dynamic_field=True)
default_schema = CollectionSchema(
fields=default_fields, description="test collection", enable_dynamic_field=True
)
# init collection by sdk
collection = Collection(name=name, schema=default_schema)
index_param = {"metric_type": "L2", "index_type": "IVF_FLAT", "params": {"nlist": 128}}
@@ -311,16 +308,13 @@ class TestRestfulSdkCompatibility(TestBase):
]
collection.insert(data)
time.sleep(5)
res = collection.query(expr=f"int64 in [0, 1, 2, 3, 4, 5, 6, 7, 8, 9]", output_fields=["*"])
res = collection.query(expr="int64 in [0, 1, 2, 3, 4, 5, 6, 7, 8, 9]", output_fields=["*"])
pk_id_list = []
for item in res:
pk_id_list.append(item["int64"])
payload = {
"collectionName": name,
"filter": f"int64 in {pk_id_list}"
}
payload = {"collectionName": name, "filter": f"int64 in {pk_id_list}"}
# delete data by restful
rsp = self.vector_client.vector_delete(payload)
self.vector_client.vector_delete(payload)
time.sleep(5)
res = collection.query(expr=f"int64 in [0, 1, 2, 3, 4, 5, 6, 7, 8, 9]", output_fields=["*"])
res = collection.query(expr="int64 in [0, 1, 2, 3, 4, 5, 6, 7, 8, 9]", output_fields=["*"])
assert len(res) == 0
@@ -1,29 +1,27 @@
from utils.utils import gen_unique_str
from base.testbase import TestBase
import pytest
from base.testbase import TestBase
from utils.constant import CaseLabel
from utils.utils import gen_unique_str
@pytest.mark.L1
@pytest.mark.tags(CaseLabel.RBAC)
class TestRoleE2E(TestBase):
def teardown_method(self):
# because role num is limited, so we need to delete all roles after test
rsp = self.role_client.role_list()
all_roles = rsp['data']
all_roles = rsp["data"]
# delete all roles except default roles
for role in all_roles:
if role.startswith("role") and role in self.role_client.role_names:
payload = {
"roleName": role
}
payload = {"roleName": role}
# revoke privilege from role
rsp = self.role_client.role_describe(role)
for d in rsp['data']:
for d in rsp["data"]:
payload = {
"roleName": role,
"objectType": d['objectType'],
"objectName": d['objectName'],
"privilege": d['privilege']
"objectType": d["objectType"],
"objectName": d["objectName"],
"privilege": d["privilege"],
}
self.role_client.role_revoke(payload)
self.role_client.role_drop(payload)
@@ -40,44 +38,31 @@ class TestRoleE2E(TestBase):
rsp = self.role_client.role_create(payload)
# list role after create
rsp = self.role_client.role_list()
assert role_name in rsp['data']
assert role_name in rsp["data"]
# describe role
rsp = self.role_client.role_describe(role_name)
assert rsp['code'] == 0
assert rsp["code"] == 0
# grant privilege to role
payload = {
"roleName": role_name,
"objectType": "Global",
"objectName": "*",
"privilege": "CreateCollection"
}
payload = {"roleName": role_name, "objectType": "Global", "objectName": "*", "privilege": "CreateCollection"}
rsp = self.role_client.role_grant(payload)
assert rsp['code'] == 0
assert rsp["code"] == 0
# describe role after grant
rsp = self.role_client.role_describe(role_name)
privileges = []
for p in rsp['data']:
privileges.append(p['privilege'])
for p in rsp["data"]:
privileges.append(p["privilege"])
assert "CreateCollection" in privileges
# revoke privilege from role
payload = {
"roleName": role_name,
"objectType": "Global",
"objectName": "*",
"privilege": "CreateCollection"
}
payload = {"roleName": role_name, "objectType": "Global", "objectName": "*", "privilege": "CreateCollection"}
rsp = self.role_client.role_revoke(payload)
# describe role after revoke
rsp = self.role_client.role_describe(role_name)
privileges = []
for p in rsp['data']:
privileges.append(p['privilege'])
for p in rsp["data"]:
privileges.append(p["privilege"])
assert "CreateCollection" not in privileges
# drop role
payload = {
"roleName": role_name
}
payload = {"roleName": role_name}
rsp = self.role_client.role_drop(payload)
rsp = self.role_client.role_list()
assert role_name not in rsp['data']
assert role_name not in rsp["data"]
@@ -1,14 +1,15 @@
import time
import random
import time
from datetime import UTC, datetime, timedelta
import pytest
from datetime import datetime, timezone, timedelta
from base.testbase import TestBase
from utils.utils import gen_collection_name
from utils.constant import CaseLabel
from utils.util_log import test_log as logger
from utils.utils import gen_collection_name
@pytest.mark.L1
@pytest.mark.tags(CaseLabel.L1)
class TestTimestamptz(TestBase):
"""
RESTful e2e coverage for timestamptz field:
@@ -59,13 +60,28 @@ class TestTimestamptz(TestBase):
assert "defaultValue" in time_field
# 3. insert rows (one row omits timestamptz to trigger default)
now_utc = datetime.now(timezone.utc)
now_utc = datetime.now(UTC)
one_hour_ago = now_utc - timedelta(hours=1)
rows = [
{"id": 1, "time": now_utc.isoformat(), "color": "red_9392", "vector": [random.random() for _ in range(dim)]},
{"id": 3, "time": one_hour_ago.isoformat(), "color": "pink_9298", "vector": [random.random() for _ in range(dim)]},
{
"id": 1,
"time": now_utc.isoformat(),
"color": "red_9392",
"vector": [random.random() for _ in range(dim)],
},
{
"id": 3,
"time": one_hour_ago.isoformat(),
"color": "pink_9298",
"vector": [random.random() for _ in range(dim)],
},
{"id": 4, "color": "green_0004", "vector": [random.random() for _ in range(dim)]}, # default timestamptz
{"id": 504, "time": one_hour_ago.isoformat(), "color": "blue_0000", "vector": [random.random() for _ in range(dim)]},
{
"id": 504,
"time": one_hour_ago.isoformat(),
"color": "blue_0000",
"vector": [random.random() for _ in range(dim)],
},
]
insert_payload = {"collectionName": name, "data": rows}
insert_rsp = self.vector_client.vector_insert(insert_payload)
@@ -103,4 +119,3 @@ class TestTimestamptz(TestBase):
assert result[1]["color"] == "red_9392"
assert result[3]["color"] == "pink_9298"
assert result[4]["color"] == "green_0004"
@@ -1,35 +1,34 @@
import time
from utils.utils import gen_collection_name, gen_unique_str
import pytest
from base.testbase import TestBase
from pymilvus import (connections)
from pymilvus import connections
from utils.constant import CaseLabel
from utils.utils import gen_collection_name, gen_unique_str
class TestUserE2E(TestBase):
def teardown_method(self):
# because role num is limited, so we need to delete all roles after test
rsp = self.role_client.role_list()
all_roles = rsp['data']
all_roles = rsp["data"]
# delete all roles except default roles
for role in all_roles:
if role.startswith("role") and role in self.role_client.role_names:
payload = {
"roleName": role
}
payload = {"roleName": role}
# revoke privilege from role
rsp = self.role_client.role_describe(role)
for d in rsp['data']:
for d in rsp["data"]:
payload = {
"roleName": role,
"objectType": d['objectType'],
"objectName": d['objectName'],
"privilege": d['privilege']
"objectType": d["objectType"],
"objectName": d["objectName"],
"privilege": d["privilege"],
}
self.role_client.role_revoke(payload)
self.role_client.role_drop(payload)
@pytest.mark.L0
@pytest.mark.tags(CaseLabel.RBAC)
def test_user_e2e(self):
# list user before create
@@ -37,48 +36,36 @@ class TestUserE2E(TestBase):
# create user
user_name = gen_unique_str("user")
password = "1234578"
payload = {
"userName": user_name,
"password": password
}
payload = {"userName": user_name, "password": password}
rsp = self.user_client.user_create(payload)
# list user after create
rsp = self.user_client.user_list()
assert user_name in rsp['data']
assert user_name in rsp["data"]
# describe user
rsp = self.user_client.user_describe(user_name)
# update user password
new_password = "87654321"
payload = {
"userName": user_name,
"password": password,
"newPassword": new_password
}
payload = {"userName": user_name, "password": password, "newPassword": new_password}
rsp = self.user_client.user_password_update(payload)
assert rsp['code'] == 0
assert rsp["code"] == 0
# drop user
payload = {
"userName": user_name
}
payload = {"userName": user_name}
rsp = self.user_client.user_drop(payload)
rsp = self.user_client.user_list()
assert user_name not in rsp['data']
assert user_name not in rsp["data"]
@pytest.mark.L1
@pytest.mark.tags(CaseLabel.RBAC)
def test_user_binding_role(self):
# create user
user_name = gen_unique_str("user")
password = "12345678"
payload = {
"userName": user_name,
"password": password
}
payload = {"userName": user_name, "password": password}
rsp = self.user_client.user_create(payload)
# list user after create
rsp = self.user_client.user_list()
assert user_name in rsp['data']
assert user_name in rsp["data"]
# create role
role_name = gen_unique_str("role")
payload = {
@@ -86,18 +73,10 @@ class TestUserE2E(TestBase):
}
rsp = self.role_client.role_create(payload)
# privilege to role
payload = {
"roleName": role_name,
"objectType": "Global",
"objectName": "*",
"privilege": "All"
}
payload = {"roleName": role_name, "objectType": "Global", "objectName": "*", "privilege": "All"}
rsp = self.role_client.role_grant(payload)
# bind role to user
payload = {
"userName": user_name,
"roleName": role_name
}
payload = {"userName": user_name, "roleName": role_name}
rsp = self.user_client.user_grant(payload)
# describe user roles
rsp = self.user_client.user_describe(user_name)
@@ -118,18 +97,17 @@ class TestUserE2E(TestBase):
{"fieldName": "book_id", "dataType": "Int64", "isPrimary": True, "elementTypeParams": {}},
{"fieldName": "word_count", "dataType": "Int64", "elementTypeParams": {}},
{"fieldName": "book_describe", "dataType": "VarChar", "elementTypeParams": {"max_length": "256"}},
{"fieldName": "book_intro", "dataType": "FloatVector", "elementTypeParams": {"dim": "128"}}
{"fieldName": "book_intro", "dataType": "FloatVector", "elementTypeParams": {"dim": "128"}},
]
}
},
}
self.collection_client.api_key = f"{user_name}:{password}"
rsp = self.collection_client.collection_create(payload)
assert rsp['code'] == 0
assert rsp["code"] == 0
@pytest.mark.L1
@pytest.mark.tags(CaseLabel.RBAC)
class TestUserNegative(TestBase):
def test_create_user_with_short_password(self):
# list user before create
@@ -137,12 +115,9 @@ class TestUserNegative(TestBase):
# create user
user_name = gen_unique_str("user")
password = "1234"
payload = {
"userName": user_name,
"password": password
}
payload = {"userName": user_name, "password": password}
rsp = self.user_client.user_create(payload)
assert rsp['code'] == 1100
assert rsp["code"] == 1100
def test_create_user_twice(self):
# list user before create
@@ -151,14 +126,11 @@ class TestUserNegative(TestBase):
# create user
user_name = gen_unique_str("user")
password = "12345678"
payload = {
"userName": user_name,
"password": password
}
payload = {"userName": user_name, "password": password}
for i in range(2):
rsp = self.user_client.user_create(payload)
if i == 0:
assert rsp['code'] == 0
assert rsp["code"] == 0
else:
assert rsp['code'] == 1100
assert "user already exists" in rsp['message']
assert rsp["code"] == 1100
assert "user already exists" in rsp["message"]
@@ -11,7 +11,7 @@ from base.testbase import TestBase
from faker import Faker
from pymilvus import Collection, CollectionSchema, DataType, FieldSchema, utility
from sklearn import preprocessing
from utils.constant import MAX_SUM_OFFSET_AND_LIMIT, default_nb
from utils.constant import MAX_SUM_OFFSET_AND_LIMIT, CaseLabel, default_nb
from utils.util_log import test_log as logger
from utils.utils import (
analyze_documents,
@@ -34,7 +34,7 @@ patch_faker_text(fake_en, en_vocabularies_distribution)
patch_faker_text(fake_zh, zh_vocabularies_distribution)
@pytest.mark.L0
@pytest.mark.tags(CaseLabel.L0)
class TestInsertVector(TestBase):
@pytest.mark.parametrize("insert_round", [3])
@pytest.mark.parametrize("nb", [3000])
@@ -927,7 +927,7 @@ class TestInsertVector(TestBase):
assert rsp["data"][0]["json"] is None
@pytest.mark.L0
@pytest.mark.tags(CaseLabel.L0)
class TestInsertVectorNegative(TestBase):
def test_insert_vector_with_invalid_collection_name(self):
"""
@@ -1158,7 +1158,7 @@ class TestInsertVectorNegative(TestBase):
assert rsp["data"]["insertCount"] == 10
@pytest.mark.L0
@pytest.mark.tags(CaseLabel.L0)
class TestUpsertVector(TestBase):
@pytest.mark.parametrize("insert_round", [2])
@pytest.mark.parametrize("nb", [3000])
@@ -1423,7 +1423,7 @@ class TestUpsertVector(TestBase):
assert data["book_describe"] is None
@pytest.mark.L0
@pytest.mark.tags(CaseLabel.L0)
class TestUpsertVectorNegative(TestBase):
def test_upsert_vector_with_invalid_collection_name(self):
"""
@@ -1553,7 +1553,7 @@ class TestSearchAggregationVector(TestBase):
assert "aggTopks" in rsp, rsp
assert "data" in rsp, rsp
@pytest.mark.L0
@pytest.mark.tags(CaseLabel.L0)
def test_search_aggregation_minimal_parameters(self):
name, _ = self._create_search_aggregation_collection()
@@ -1578,7 +1578,7 @@ class TestSearchAggregationVector(TestBase):
assert buckets[0]["metrics"] == {}
assert buckets[0]["hits"] == []
@pytest.mark.L0
@pytest.mark.tags(CaseLabel.L0)
def test_search_aggregation(self):
name = gen_collection_name()
payload = {
@@ -1743,7 +1743,7 @@ class TestSearchAggregationVector(TestBase):
assert rsp["code"] != 0
assert "searchAggregation is not supported for hybrid search" in rsp["message"]
@pytest.mark.L1
@pytest.mark.tags(CaseLabel.L1)
def test_search_aggregation_composite_key_metrics_order(self):
name, data = self._create_search_aggregation_collection()
@@ -1790,7 +1790,7 @@ class TestSearchAggregationVector(TestBase):
assert hit["brand"] == brand
assert hit["color"] == color
@pytest.mark.L1
@pytest.mark.tags(CaseLabel.L1)
def test_search_aggregation_two_level_nested_with_filter(self):
name, data = self._create_search_aggregation_collection()
filtered = [row for row in data if row["in_stock"]]
@@ -1876,7 +1876,7 @@ class TestSearchAggregationVector(TestBase):
),
],
)
@pytest.mark.L1
@pytest.mark.tags(CaseLabel.L1)
def test_search_aggregation_reject_invalid_parameters(self, search_aggregation, message):
name, _ = self._create_search_aggregation_collection()
@@ -1892,7 +1892,7 @@ class TestSearchAggregationVector(TestBase):
assert rsp["code"] != 0
assert message in rsp["message"]
@pytest.mark.L1
@pytest.mark.tags(CaseLabel.L1)
@pytest.mark.parametrize(
"search_aggregation,message",
[
@@ -1906,7 +1906,7 @@ class TestSearchAggregationVector(TestBase):
),
(
{"fields": ["brand"], "size": 1, "order": [{"key": "_count", "direction": "up"}]},
"up",
"asc or desc",
),
],
)
@@ -1926,7 +1926,7 @@ class TestSearchAggregationVector(TestBase):
assert message in rsp["message"]
@pytest.mark.L0
@pytest.mark.tags(CaseLabel.L0)
class TestSearchVector(TestBase):
@pytest.mark.parametrize("insert_round", [1])
@pytest.mark.parametrize("auto_id", [True])
@@ -3285,7 +3285,7 @@ class TestSearchVector(TestBase):
assert len(rsp["data"]) == 100 * nq
@pytest.mark.L0
@pytest.mark.tags(CaseLabel.L0)
class TestSearchVectorNegative(TestBase):
@pytest.mark.parametrize("metric_type", ["L2"])
def test_search_vector_without_required_data_param(self, metric_type):
@@ -3345,7 +3345,7 @@ class TestSearchVectorNegative(TestBase):
"offset": 0,
}
rsp = self.vector_client.vector_search(payload)
assert rsp['code'] == 1100
assert rsp["code"] == 1100
@pytest.mark.parametrize("offset", [-1, 100_001])
def test_search_vector_with_invalid_offset(self, offset):
@@ -3370,7 +3370,7 @@ class TestSearchVectorNegative(TestBase):
"offset": offset,
}
rsp = self.vector_client.vector_search(payload)
assert rsp['code'] == 1100
assert rsp["code"] == 1100
def test_search_vector_with_invalid_collection_name(self):
"""
@@ -3397,7 +3397,7 @@ class TestSearchVectorNegative(TestBase):
assert "can't find collection" in rsp["message"]
@pytest.mark.L0
@pytest.mark.tags(CaseLabel.L0)
class TestAdvancedSearchVector(TestBase):
@pytest.mark.parametrize("insert_round", [1])
@pytest.mark.parametrize("auto_id", [True])
@@ -3508,7 +3508,7 @@ class TestAdvancedSearchVector(TestBase):
assert rsp["topks"][0] == 10
@pytest.mark.L0
@pytest.mark.tags(CaseLabel.L0)
class TestHybridSearchVector(TestBase):
@pytest.mark.parametrize("insert_round", [1])
@pytest.mark.parametrize("auto_id", [True])
@@ -4271,8 +4271,8 @@ class TestHybridSearchVector(TestBase):
rsp = self.vector_client.vector_hybrid_search(payload)
# When offset + limit exceeds max allowed
if large_offset + limit > MAX_SUM_OFFSET_AND_LIMIT:
assert rsp['code'] == 1100
assert "exceeds" in rsp['message'] or "invalid" in rsp['message'].lower()
assert rsp["code"] == 1100
assert "exceeds" in rsp["message"] or "invalid" in rsp["message"].lower()
# When offset is larger than the available results
elif large_offset >= nb:
# Should return empty results or handle gracefully
@@ -4286,7 +4286,7 @@ class TestHybridSearchVector(TestBase):
assert len(rsp["data"]) == expected_count
@pytest.mark.L0
@pytest.mark.tags(CaseLabel.L0)
class TestQueryVector(TestBase):
@pytest.mark.parametrize("insert_round", [1])
@pytest.mark.parametrize("auto_id", [True])
@@ -5064,7 +5064,7 @@ class TestQueryVector(TestBase):
assert len(rsp["data"]) == 50
@pytest.mark.L0
@pytest.mark.tags(CaseLabel.L0)
class TestQueryVectorNegative(TestBase):
def test_query_with_wrong_filter_expr(self):
name = gen_collection_name()
@@ -5086,7 +5086,7 @@ class TestQueryVectorNegative(TestBase):
assert "failed to create query plan" in rsp["message"]
@pytest.mark.L0
@pytest.mark.tags(CaseLabel.L0)
class TestGetVector(TestBase):
def test_get_vector_with_simple_payload(self):
"""
@@ -5124,7 +5124,7 @@ class TestGetVector(TestBase):
for item in res:
assert item["id"] == ids[0]
@pytest.mark.L0
@pytest.mark.tags(CaseLabel.L0)
@pytest.mark.parametrize("id_field_type", ["list", "one"])
@pytest.mark.parametrize("include_invalid_id", [True, False])
@pytest.mark.parametrize("include_output_fields", [True, False])
@@ -5186,7 +5186,7 @@ class TestGetVector(TestBase):
assert field in r
@pytest.mark.L0
@pytest.mark.tags(CaseLabel.L0)
class TestDeleteVector(TestBase):
@pytest.mark.xfail(reason="delete by id is not supported")
def test_delete_vector_by_id(self):
@@ -5459,7 +5459,7 @@ class TestDeleteVector(TestBase):
assert len(rsp["data"]) == 0
@pytest.mark.L0
@pytest.mark.tags(CaseLabel.L0)
class TestDeleteVectorNegative(TestBase):
def test_delete_vector_with_invalid_collection_name(self):
"""
@@ -5501,7 +5501,7 @@ class TestDeleteVectorNegative(TestBase):
assert "can't find collection" in rsp["message"]
@pytest.mark.L1
@pytest.mark.tags(CaseLabel.RBAC)
class TestVectorWithAuth(TestBase):
def test_upsert_vector_with_invalid_api_key(self):
"""
@@ -5605,7 +5605,7 @@ class TestVectorWithAuth(TestBase):
assert rsp["code"] == 1800
@pytest.mark.L0
@pytest.mark.tags(CaseLabel.L0)
class TestSearchByPK(TestBase):
"""
Test cases for search by primary key functionality (PR #47260)
@@ -6098,7 +6098,7 @@ class TestSearchByPK(TestBase):
assert pk in returned_pks
@pytest.mark.L0
@pytest.mark.tags(CaseLabel.L0)
class TestSearchByPKNegative(TestBase):
"""Negative test cases for search by primary key functionality"""
+8 -1
View File
@@ -1,3 +1,10 @@
default_nb = 3000
MAX_SUM_OFFSET_AND_LIMIT = 16384
class CaseLabel:
L0 = "L0"
L1 = "L1"
L2 = "L2"
L3 = "L3"
RBAC = "RBAC"
+5 -4
View File
@@ -16,8 +16,9 @@ class TestLog:
self.log.setLevel(logging.DEBUG)
try:
formatter = logging.Formatter("[%(asctime)s - %(levelname)s - %(name)s]: "
"%(message)s (%(filename)s:%(lineno)s)")
formatter = logging.Formatter(
"[%(asctime)s - %(levelname)s - %(name)s]: %(message)s (%(filename)s:%(lineno)s)"
)
# [%(process)s] process NO.
dh = logging.FileHandler(self.log_debug)
dh.setLevel(logging.DEBUG)
@@ -45,7 +46,7 @@ class TestLog:
ch.setFormatter(formatter)
except Exception as e:
print("Can not use %s or %s or %s to log. error : %s" % (log_debug, log_file, log_err, str(e)))
print(f"Can not use {log_debug} or {log_file} or {log_err} to log. error : {e}")
"""All modules share this unified log"""
@@ -53,4 +54,4 @@ log_debug = log_config.log_debug
log_info = log_config.log_info
log_err = log_config.log_err
log_worker = log_config.log_worker
test_log = TestLog('ci_test', log_debug, log_info, log_err, log_worker).log
test_log = TestLog("ci_test", log_debug, log_info, log_err, log_worker).log
+2 -11
View File
@@ -84,18 +84,9 @@ cd ${ROOT}/tests/restful_client_v2
if [[ -n "${TEST_TIMEOUT:-}" ]]; then
timeout "${TEST_TIMEOUT}" pytest testcases --endpoint http://${MILVUS_SERVICE_NAME}:${MILVUS_SERVICE_PORT} --minio_host ${MINIO_SERVICE_NAME} -v -x -m L0 -n 6 --timeout 360
timeout "${TEST_TIMEOUT}" pytest testcases --endpoint http://${MILVUS_SERVICE_NAME}:${MILVUS_SERVICE_PORT} --minio_host ${MINIO_SERVICE_NAME} -v -x --tags L0 L1 -n 6 --timeout 360
else
pytest testcases --endpoint http://${MILVUS_SERVICE_NAME}:${MILVUS_SERVICE_PORT} --minio_host ${MINIO_SERVICE_NAME} -v -x -m L0 -n 6 --timeout 360
fi
if [[ "${MILVUS_HELM_RELEASE_NAME}" != *"msop"* ]]; then
if [[ -n "${TEST_TIMEOUT:-}" ]]; then
timeout "${TEST_TIMEOUT}" pytest testcases --endpoint http://${MILVUS_SERVICE_NAME}:${MILVUS_SERVICE_PORT} --minio_host ${MINIO_SERVICE_NAME} -v -x -m BulkInsert -n 6 --timeout 360
else
pytest testcases --endpoint http://${MILVUS_SERVICE_NAME}:${MILVUS_SERVICE_PORT} --minio_host ${MINIO_SERVICE_NAME} -v -x -m BulkInsert -n 6 --timeout 360
fi
pytest testcases --endpoint http://${MILVUS_SERVICE_NAME}:${MILVUS_SERVICE_PORT} --minio_host ${MINIO_SERVICE_NAME} -v -x --tags L0 L1 -n 6 --timeout 360
fi
cd ${ROOT}/tests/python_client
+2 -11
View File
@@ -84,18 +84,9 @@ cd ${ROOT}/tests/restful_client_v2
if [[ -n "${TEST_TIMEOUT:-}" ]]; then
timeout "${TEST_TIMEOUT}" pytest testcases --endpoint http://${MILVUS_SERVICE_NAME}:${MILVUS_SERVICE_PORT} --minio_host ${MINIO_SERVICE_NAME} -v -x -m L0 -n 6 --timeout 360
timeout "${TEST_TIMEOUT}" pytest testcases --endpoint http://${MILVUS_SERVICE_NAME}:${MILVUS_SERVICE_PORT} --minio_host ${MINIO_SERVICE_NAME} -v -x --tags L0 L1 -n 6 --timeout 360
else
pytest testcases --endpoint http://${MILVUS_SERVICE_NAME}:${MILVUS_SERVICE_PORT} --minio_host ${MINIO_SERVICE_NAME} -v -x -m L0 -n 6 --timeout 360
fi
if [[ "${MILVUS_HELM_RELEASE_NAME}" != *"msop"* ]]; then
if [[ -n "${TEST_TIMEOUT:-}" ]]; then
timeout "${TEST_TIMEOUT}" pytest testcases --endpoint http://${MILVUS_SERVICE_NAME}:${MILVUS_SERVICE_PORT} --minio_host ${MINIO_SERVICE_NAME} -v -x -m BulkInsert -n 6 --timeout 360
else
pytest testcases --endpoint http://${MILVUS_SERVICE_NAME}:${MILVUS_SERVICE_PORT} --minio_host ${MINIO_SERVICE_NAME} -v -x -m BulkInsert -n 6 --timeout 360
fi
pytest testcases --endpoint http://${MILVUS_SERVICE_NAME}:${MILVUS_SERVICE_PORT} --minio_host ${MINIO_SERVICE_NAME} -v -x --tags L0 L1 -n 6 --timeout 360
fi
cd ${ROOT}/tests/python_client