milvus/tests/python_client/conftest.py

import functools
import logging
import os
import socket

import common.common_func as cf
import common.common_type as ct
import numpy as np
import pytest
from check.param_check import ip_check, number_check
from common.common_func import param_info
from config.log_config import log_config
from pymilvus.orm.types import CONSISTENCY_STRONG
from utils.util_log import test_log as log
from utils.util_pymilvus import gen_binary_default_fields, gen_default_fields, gen_unique_str, get_milvus

# Legacy wrappers still import numpy.NaN; NumPy 2 only exposes numpy.nan.
if not hasattr(np, "NaN"):
    np.NaN = np.nan

# Register the log filter plugin
pytest_plugins = ["plugin.log_filter"]

timeout = 60
dimension = 128
delete_timeout = 60


def pytest_addoption(parser):
    parser.addoption("--host", action="store", default="localhost", help="service's ip")
    parser.addoption("--service", action="store", default="", help="service address")
    parser.addoption("--port", action="store", default=19530, help="service's port")
    parser.addoption("--user", action="store", default="root", help="user name for connection")
    parser.addoption("--password", action="store", default="Milvus", help="password for connection")
    parser.addoption("--db_name", action="store", default="default", help="database name for connection")
    parser.addoption("--secure", action="store", default=False, help="secure for connection")
    parser.addoption(
        "--milvus_ns",
        action="store",
        default=os.getenv("MILVUS_NS", os.getenv("MILVUS_HELM_NAMESPACE", "chaos-testing")),
        help="milvus_ns",
    )
    parser.addoption("--http_port", action="store", default=19121, help="http's port")
    parser.addoption("--handler", action="store", default="GRPC", help="handler of request")
    parser.addoption("--tag", action="store", default="all", help="only run tests matching the tag.")
    parser.addoption("--dry_run", action="store_true", default=False, help="")
    parser.addoption(
        "--run-compaction-integrity-serial",
        action="store_true",
        default=False,
        help="run cluster-global compaction data-integrity workloads in a non-xdist stage",
    )
    parser.addoption("--database_name", action="store", default="default", help="name of database")
    parser.addoption("--partition_name", action="store", default="partition_name", help="name of partition")
    parser.addoption("--connect_name", action="store", default="connect_name", help="name of connect")
    parser.addoption("--descriptions", action="store", default="partition_des", help="descriptions of partition")
    parser.addoption("--collection_name", action="store", default="collection_name", help="name of collection")
    parser.addoption("--search_vectors", action="store", default="search_vectors", help="vectors of search")
    parser.addoption("--index_param", action="store", default="index_param", help="index_param of index")
    parser.addoption("--data", action="store", default="data", help="data of request")
    parser.addoption("--clean_log", action="store_true", default=True, help="clean log before testing (default: True)")
    parser.addoption("--schema", action="store", default="schema", help="schema of test interface")
    parser.addoption("--err_msg", action="store", default="err_msg", help="error message of test")
    parser.addoption("--term_expr", action="store", default="term_expr", help="expr of query quest")
    parser.addoption("--check_content", action="store", default="check_content", help="content of check")
    parser.addoption("--field_name", action="store", default="field_name", help="field_name of index")
    parser.addoption("--replica_num", action="store", default=ct.default_replica_num, help="memory replica number")
    parser.addoption("--minio_host", action="store", default="localhost", help="minio service's ip")
    parser.addoption("--minio_bucket", action="store", default="milvus-bucket", help="minio bucket name")
    parser.addoption("--etcd_host", action="store", default="localhost", help="etcd service host")
    parser.addoption("--etcd_port", action="store", type=int, default=2379, help="etcd service port")
    parser.addoption("--etcd_root_path", action="store", default="by-dev", help="Milvus root path in etcd")
    parser.addoption("--etcd_user", action="store", default="", help="etcd authentication user")
    parser.addoption("--etcd_password", action="store", default="", help="etcd authentication password")
    parser.addoption("--uri", action="store", default="", help="uri for milvus client")
    parser.addoption("--token", action="store", default="root:Milvus", help="token for milvus client")
    parser.addoption("--request_duration", action="store", default="10m", help="request_duration")
    parser.addoption("--data_size", type=int, action="store", default=3000, help="data size for deploy test")
    parser.addoption("--is_check", action="store", type=bool, default=False, help="is_check")
    # 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 embedding endpoint",
    )

    parser.addoption(
        "--tei_reranker_endpoint",
        action="store",
        default="http://text-rerank-service.milvus-ci.svc.cluster.local:80",
        help="tei rerank endpoint",
    )
    parser.addoption(
        "--vllm_reranker_endpoint",
        action="store",
        default="http://vllm-rerank-service.milvus-ci.svc.cluster.local:80",
        help="vllm rerank endpoint",
    )
    # L3 test options for alter function tests
    parser.addoption(
        "--tei_endpoint_2", action="store", default="", help="second tei embedding endpoint for alter tests"
    )

    spark_backfill = parser.getgroup("spark-backfill")
    spark_backfill.addoption(
        "--run-spark-backfill",
        action="store_true",
        default=False,
        help="collect and run the Spark-Milvus Backfill suite",
    )
    spark_backfill.addoption(
        "--spark-runner-mode",
        action="store",
        choices=("job", "toolbox"),
        default="job",
        help="run Spark in one-shot Kubernetes Jobs or an existing Toolbox Pod",
    )
    spark_backfill.addoption("--management-endpoint", action="store", default="")
    spark_backfill.addoption("--spark-k8s-context", action="store", default="")
    spark_backfill.addoption("--spark-k8s-namespace", action="store", default="")
    spark_backfill.addoption(
        "--spark-image",
        action="store",
        default=(
            "apache/spark:4.0.1-scala2.13-java21-python3-ubuntu"
            "@sha256:fb5c5e61e7bb1be94b7f3a31afe1f73c5b4d20b6008f4ffa7278fc085da08a9e"
        ),
    )
    spark_backfill.addoption("--spark-connector-url", action="store", default="")
    spark_backfill.addoption("--spark-connector-sha256", action="store", default="")
    spark_backfill.addoption("--spark-toolbox-pod", action="store", default="")
    spark_backfill.addoption(
        "--spark-toolbox-label",
        action="store",
        default="app=spark-milvus-toolbox",
    )
    spark_backfill.addoption("--spark-toolbox-container", action="store", default="spark-toolbox")
    spark_backfill.addoption(
        "--spark-toolbox-wrapper",
        action="store",
        default="/usr/local/bin/spark-submit-milvus",
    )
    spark_backfill.addoption(
        "--spark-toolbox-workspace",
        action="store",
        default="/workspace/spark-backfill-pytest",
    )
    spark_backfill.addoption("--spark-milvus-uri", action="store", default="")
    spark_backfill.addoption("--spark-minio-endpoint", action="store", default="")
    spark_backfill.addoption("--spark-storage-secret-name", action="store", default="")
    spark_backfill.addoption("--spark-service-account-name", action="store", default="")
    spark_backfill.addoption("--spark-job-timeout", action="store", type=int, default=1800)
    spark_backfill.addoption("--spark-keep-failed-job", action="store_true", default=False)
    spark_backfill.addoption(
        "--spark-evidence-root",
        action="store",
        default=os.path.join(os.getenv("CI_LOG_PATH", "/tmp/ci_logs"), "spark_backfill"),
    )


def pytest_ignore_collect(collection_path, config):
    """Keep Spark Backfill tests completely out of regular PR/E2E collection."""
    from spark_backfill.config import is_spark_backfill_path

    if is_spark_backfill_path(collection_path) and not config.getoption("--run-spark-backfill"):
        return True
    return None


@pytest.fixture
def host(request):
    return request.config.getoption("--host")


@pytest.fixture
def service(request):
    return request.config.getoption("--service")


@pytest.fixture
def port(request):
    return request.config.getoption("--port")


@pytest.fixture
def user(request):
    return request.config.getoption("--user")


@pytest.fixture
def password(request):
    return request.config.getoption("--password")


@pytest.fixture
def db_name(request):
    return request.config.getoption("--db_name")


@pytest.fixture
def secure(request):
    return request.config.getoption("--secure")


@pytest.fixture
def milvus_ns(request):
    return request.config.getoption("--milvus_ns")


@pytest.fixture
def http_port(request):
    return request.config.getoption("--http_port")


@pytest.fixture
def handler(request):
    return request.config.getoption("--handler")


@pytest.fixture
def tag(request):
    return request.config.getoption("--tag")


@pytest.fixture
def dry_run(request):
    return request.config.getoption("--dry_run")


@pytest.fixture
def connect_name(request):
    return request.config.getoption("--connect_name")


@pytest.fixture
def database_name(request):
    return request.config.getoption("--database_name")


@pytest.fixture
def partition_name(request):
    return request.config.getoption("--partition_name")


@pytest.fixture
def descriptions(request):
    return request.config.getoption("--descriptions")


@pytest.fixture
def collection_name(request):
    return request.config.getoption("--collection_name")


@pytest.fixture
def search_vectors(request):
    return request.config.getoption("--search_vectors")


@pytest.fixture
def index_param(request):
    return request.config.getoption("--index_param")


@pytest.fixture
def data(request):
    return request.config.getoption("--data")


@pytest.fixture
def clean_log(request):
    return request.config.getoption("--clean_log")


@pytest.fixture
def schema(request):
    return request.config.getoption("--schema")


@pytest.fixture
def err_msg(request):
    return request.config.getoption("--err_msg")


@pytest.fixture
def term_expr(request):
    return request.config.getoption("--term_expr")


@pytest.fixture
def check_content(request):
    log.error("^" * 50)
    log.error("check_content")
    return request.config.getoption("--check_content")


@pytest.fixture
def field_name(request):
    return request.config.getoption("--field_name")


@pytest.fixture
def minio_host(request):
    return request.config.getoption("--minio_host")


@pytest.fixture
def minio_bucket(request):
    return request.config.getoption("--minio_bucket")


@pytest.fixture
def etcd_host(request):
    return request.config.getoption("--etcd_host")


@pytest.fixture
def etcd_port(request):
    return request.config.getoption("--etcd_port")


@pytest.fixture
def etcd_root_path(request):
    return request.config.getoption("--etcd_root_path")


@pytest.fixture
def etcd_user(request):
    return request.config.getoption("--etcd_user")


@pytest.fixture
def etcd_password(request):
    return request.config.getoption("--etcd_password")


@pytest.fixture
def uri(request):
    return request.config.getoption("--uri")


@pytest.fixture
def token(request):
    return request.config.getoption("--token")


@pytest.fixture
def request_duration(request):
    return request.config.getoption("--request_duration")


@pytest.fixture
def tei_endpoint(request):
    return request.config.getoption("--tei_endpoint")


@pytest.fixture
def tei_endpoint_2(request):
    endpoint = request.config.getoption("--tei_endpoint_2")
    if not endpoint:
        pytest.skip("tei_endpoint_2 not configured")
    return endpoint


@pytest.fixture
def tei_reranker_endpoint(request):
    return request.config.getoption("--tei_reranker_endpoint")


@pytest.fixture
def vllm_reranker_endpoint(request):
    return request.config.getoption("--vllm_reranker_endpoint")


@pytest.fixture
def data_size(request):
    return request.config.getoption("--data_size")


@pytest.fixture
def is_check(request):
    return request.config.getoption("--is_check")


""" fixture func """


@pytest.fixture(scope="session", autouse=True)
def initialize_env(request):
    """clean log before testing"""
    host = request.config.getoption("--host")
    port = request.config.getoption("--port")
    handler = request.config.getoption("--handler")
    user = request.config.getoption("--user")
    password = request.config.getoption("--password")
    secure = request.config.getoption("--secure")
    clean_log = request.config.getoption("--clean_log")
    replica_num = request.config.getoption("--replica_num")
    uri = request.config.getoption("--uri")
    token = request.config.getoption("--token")
    minio_bucket = request.config.getoption("--minio_bucket")

    """ params check """
    assert ip_check(host) and number_check(port)

    """ modify log files """
    file_path_list = [log_config.log_report_json, log_config.log_report_html]
    cf.modify_file(file_path_list=file_path_list, is_modify=clean_log)

    log.info("#" * 80)
    log.info("[initialize_milvus] Log cleaned up, start testing...")
    param_info.prepare_param_info(host, port, handler, replica_num, user, password, secure, uri, token, minio_bucket)


# TODO: construct invalid index params for all index types
@pytest.fixture(
    params=[
        {"metric_type": "L3", "index_type": "IVF_FLAT"},
        {"metric_type": "L2", "index_type": "IVF_FLAT", "params": {"nlist": -1}},
    ]
)
def get_invalid_index_params(request):
    yield request.param


@pytest.fixture(params=ct.get_invalid_dict)
def get_invalid_vector_dict(request):
    yield request.param


def pytest_configure(config):
    # register an additional marker
    config.addinivalue_line("markers", "tag(name): mark test to run only matching the tag")
    if config.getoption("--run-spark-backfill"):
        from spark_backfill.config import ensure_serial_execution

        ensure_serial_execution(getattr(config.option, "numprocesses", None))


def pytest_runtest_setup(item):
    if item.get_closest_marker("compaction_data_integrity_serial") is not None:
        if not item.config.getoption("--run-compaction-integrity-serial"):
            pytest.skip("compaction data-integrity workloads require the dedicated serial E2E stage")
        if hasattr(item.config, "workerinput"):
            pytest.fail("compaction data-integrity workloads must run with pytest -n 0")
    tags = list()
    for marker in item.iter_markers(name="tag"):
        for tag in marker.args:
            tags.append(tag)
    if tags:
        cmd_tag = item.config.getoption("--tag")
        if cmd_tag != "all" and cmd_tag not in tags:
            pytest.skip(f"test requires tag in {tags!r}")


def pytest_runtestloop(session):
    if session.config.getoption("--dry_run"):
        total_num = 0
        tags_num = 0
        res = {"total_num": total_num, "tags_num": tags_num}
        for item in session.items:
            print(item.nodeid)
            if item.fspath.basename not in res:
                res.update({item.fspath.basename: {"total": 1, "tags": 0}})
            else:
                res[item.fspath.basename]["total"] += 1
            res["total_num"] += 1
            for marker in item.own_markers:
                if marker.name == "tags" and "0331" in marker.args:
                    res["tags_num"] += 1
                    res[item.fspath.basename]["tags"] += 1
        print(res)
        return True


def check_server_connection(request):
    host = request.config.getoption("--host")
    port = request.config.getoption("--port")

    connected = True
    if host and (host not in ["localhost", "127.0.0.1"]):
        try:
            socket.getaddrinfo(host, port, 0, 0, socket.IPPROTO_TCP)
        except Exception as e:
            print(f"Socket connnet failed: {e!s}")
            connected = False
    return connected


# @pytest.fixture(scope="session", autouse=True)
# def change_mutation_result_to_primary_keys():
#     def insert_future_decorator(func):
#         @functools.wraps(func)
#         def change(*args, **kwargs):
#             try:
#                 return func(*args, **kwargs).primary_keys
#             except Exception as e:
#                 raise e
#         return change
#
#     from pymilvus import MutationFuture
#     MutationFuture.result = insert_future_decorator(MutationFuture.result)
#
#     def insert_decorator(func):
#         @functools.wraps(func)
#         def change(*args, **kwargs):
#             if kwargs.get("_async", False):
#                 return func(*args, **kwargs)
#             try:
#                 return func(*args, **kwargs).primary_keys
#             except Exception as e:
#                 raise e
#         return change
#     Milvus.insert = insert_decorator(Milvus.insert)
#     yield


# @pytest.fixture(scope="module")
def connect(request):
    host = request.config.getoption("--host")
    port = request.config.getoption("--port")
    http_port = request.config.getoption("--http_port")
    handler = request.config.getoption("--handler")
    if handler == "HTTP":
        port = http_port
    try:
        milvus = get_milvus(host=host, port=port, handler=handler)
        # reset_build_index_threshold(milvus)
    except Exception as e:
        logging.getLogger().error(str(e))
        pytest.exit("Milvus server can not connected, exit pytest ...")

    def fin():
        try:
            milvus.close()
            pass
        except Exception as e:
            logging.getLogger().info(str(e))

    request.addfinalizer(fin)
    return milvus


# @pytest.fixture(scope="module")
def dis_connect(request):
    host = request.config.getoption("--host")
    port = request.config.getoption("--port")
    http_port = request.config.getoption("--http_port")
    handler = request.config.getoption("--handler")
    if handler == "HTTP":
        port = http_port
    milvus = get_milvus(host=host, port=port, handler=handler)
    milvus.close()
    return milvus


@pytest.fixture(scope="module")
def args(request):
    host = request.config.getoption("--host")
    service_name = request.config.getoption("--service")
    port = request.config.getoption("--port")
    http_port = request.config.getoption("--http_port")
    handler = request.config.getoption("--handler")
    if handler == "HTTP":
        port = http_port
    args = {"ip": host, "port": port, "handler": handler, "service_name": service_name}
    return args


@pytest.fixture(scope="module")
def milvus(request):
    host = request.config.getoption("--host")
    port = request.config.getoption("--port")
    http_port = request.config.getoption("--http_port")
    handler = request.config.getoption("--handler")
    if handler == "HTTP":
        port = http_port
    return get_milvus(host=host, port=port, handler=handler)


@pytest.fixture(scope="function")
def collection(request, connect):
    ori_collection_name = getattr(request.module, "collection_id", "test")
    collection_name = gen_unique_str(ori_collection_name)
    log.debug(f"collection_name: {collection_name}")
    try:
        default_fields = gen_default_fields()
        connect.create_collection(collection_name, default_fields, consistency_level=CONSISTENCY_STRONG)
    except Exception as e:
        pytest.exit(str(e))

    def teardown():
        if connect.has_collection(collection_name):
            connect.drop_collection(collection_name, timeout=delete_timeout)

    request.addfinalizer(teardown)
    assert connect.has_collection(collection_name)
    return collection_name


# customised id
@pytest.fixture(scope="function")
def id_collection(request, connect):
    ori_collection_name = getattr(request.module, "collection_id", "test")
    collection_name = gen_unique_str(ori_collection_name)
    log.debug(f"id_collection_name: {collection_name}")
    try:
        fields = gen_default_fields(auto_id=False)
        connect.create_collection(collection_name, fields, consistency_level=CONSISTENCY_STRONG)
    except Exception as e:
        pytest.exit(str(e))

    def teardown():
        if connect.has_collection(collection_name):
            connect.drop_collection(collection_name, timeout=delete_timeout)

    request.addfinalizer(teardown)
    assert connect.has_collection(collection_name)
    return collection_name


@pytest.fixture(scope="function")
def binary_collection(request, connect):
    ori_collection_name = getattr(request.module, "collection_id", "test")
    collection_name = gen_unique_str(ori_collection_name)
    try:
        fields = gen_binary_default_fields()
        connect.create_collection(collection_name, fields, consistency_level=CONSISTENCY_STRONG)
    except Exception as e:
        pytest.exit(str(e))

    def teardown():
        if connect.has_collection(collection_name):
            connect.drop_collection(collection_name, timeout=delete_timeout)

    request.addfinalizer(teardown)
    assert connect.has_collection(collection_name)
    return collection_name


# customised id
@pytest.fixture(scope="function")
def binary_id_collection(request, connect):
    ori_collection_name = getattr(request.module, "collection_id", "test")
    collection_name = gen_unique_str(ori_collection_name)
    try:
        fields = gen_binary_default_fields(auto_id=False)
        connect.create_collection(collection_name, fields, consistency_level=CONSISTENCY_STRONG)
    except Exception as e:
        pytest.exit(str(e))

    def teardown():
        if connect.has_collection(collection_name):
            connect.drop_collection(collection_name, timeout=delete_timeout)

    request.addfinalizer(teardown)
    assert connect.has_collection(collection_name)
    return collection_name


# for test exit in the future
# @pytest.hookimpl(hookwrapper=True, tryfirst=True)
# def pytest_runtest_makereport():
#     result = yield
#     report = result.get_result()
#     if report.outcome == "failed":
#         msg = "The execution of the test case fails and the test exits..."
#         log.error(msg)
#         pytest.exit(msg)


# ---------------------------------------------------------------------------
# FileResource test fixtures
# ---------------------------------------------------------------------------

# Test data content generated at runtime (no files committed to git)
_FILE_RESOURCE_DATA = {
    "jieba/jieba_dict.txt": "向量数据库 5 n\n语义搜索 5 n\n全文检索 5 n\n",
    "synonyms/synonyms.txt": "向量, 矢量, vector\n搜索, 检索, 查询\n数据库, DB\n",
    "stopwords/stop_words.txt": "的\n是\n在\n了\n和\n",
    "decompounder/decompounder_dict.txt": "bank\nnote\nfire\nwork\n",
}


def _generate_file_resource_testdata(tmpdir):
    """Generate test data files in a temporary directory, preserving the
    remote directory structure so that file paths are usable with
    ``copy_files_to_minio``.  Returns a list of relative paths."""
    rel_paths = []
    for remote_path, content in _FILE_RESOURCE_DATA.items():
        local_path = os.path.join(tmpdir, remote_path)
        os.makedirs(os.path.dirname(local_path), exist_ok=True)
        with open(local_path, "w", encoding="utf-8") as f:
            f.write(content)
        rel_paths.append(remote_path)
    return rel_paths


@pytest.fixture(scope="module")
def file_resource_env(request, tmp_path_factory):
    """Generate testdata, upload to MinIO for file resource tests.

    Files are uploaded to the bucket root (no rootPath prefix) because
    AddFileResource's Exist check uses RemoteChunkManager which does NOT
    prepend minio.rootPath.

    Returns a dict with keys: bucket, minio_endpoint.
    """
    from common.minio_comm import copy_files_to_minio

    minio_host = request.config.getoption("--minio_host")
    bucket = request.config.getoption("--minio_bucket")
    tmpdir = str(tmp_path_factory.mktemp("file_resource_testdata"))
    rel_paths = _generate_file_resource_testdata(tmpdir)
    minio_endpoint = f"{minio_host}:9000"
    copy_files_to_minio(
        host=minio_endpoint,
        r_source=tmpdir,
        files=rel_paths,
        bucket_name=bucket,
        force=True,
    )

    yield {"bucket": bucket, "minio_endpoint": minio_endpoint}