test_mysql

Complexity

Total Complexity

3

Size/Duplication

Total Lines	54
Duplicated Lines	0 %

Importance

Changes

0

2 Methods

Rating	Name	Duplication	Size	Complexity
A	TestMysql.skip_check()	0	6	2
A	TestMysql._test_kill_mysql_during_index()	0	21	1

import time
import random
import pdb
import threading
import logging
from multiprocessing import Pool, Process
import pytest
from milvus import IndexType, MetricType
from utils import *


dim = 128
index_file_size = 10
collection_id = "mysql_failure"
nprobe = 1
tag = "1970-01-01"


class TestMysql:

    """
    ******************************************************************
      The following cases are used to test mysql failure
    ******************************************************************
    """
    @pytest.fixture(scope="function", autouse=True)
    def skip_check(self, connect, args):
        if args["service_name"].find("shards") != -1:
            reason = "Skip restart cases in shards mode"
            logging.getLogger().info(reason)
            pytest.skip(reason)

    def _test_kill_mysql_during_index(self, connect, collection, args):
        big_nb = 20000
        index_param = {"nlist": 1024, "m": 16}
        index_type = IndexType.IVF_PQ
        vectors = gen_vectors(big_nb, dim)

The variable gen_vectors does not seem to be defined.

        status, ids = connect.insert(collection, vectors, ids=[i for i in range(big_nb)])
        status = connect.flush([collection])
        assert status.OK()
        status, res_count = connect.count_entities(collection)
        logging.getLogger().info(res_count)
        assert status.OK()
        assert res_count == big_nb
        logging.getLogger().info("Start create index async")
        status = connect.create_index(collection, index_type, index_param, _async=True)
        time.sleep(2)
        logging.getLogger().info("Start play mysql failure")
        # pass
        new_connect = get_milvus(args["ip"], args["port"], handler=args["handler"])

The variable get_milvus does not seem to be defined.

        status, res_count = new_connect.count_entities(collection)
        assert status.OK()
        assert res_count == big_nb