def test_search_index_refresh_on_interval_multi_threaded(self):
        """ Same as test_index_refresh_on_interval_multi_threaded() ,
        but using the search endpoint.

        The same caveat applies: Because checking the last_refresh_time
        isn't threadsafe, this test may occasionally fail.
        """

        mock_get = mock.Mock()
        mock_response = requests.Response()
        mock_response.status_code = 200
        mock_response.json = lambda: '{"a":"b"}'
        mock_get.return_value = mock_response

        # we need to search it once, to get something in the cache, otherwise
        # the threads will see an empty cache and try to fill it
        try:
            self.add_documents(
                config=self.config, add_docs_params=AddDocsParams(
                    index_name=self.index_name_1, docs=[{"hi": "hello"}],
                    auto_refresh=False, device="cpu"))
        except IndexNotFoundError:
            pass
        @mock.patch('marqo._httprequests.ALLOWED_OPERATIONS', {mock_get})
        @mock.patch('marqo._httprequests.requests.get', mock_get)
        def run():

            # requests.get('23456')
            N_seconds = 4
            # the following is hard coded in search()
            REFRESH_INTERVAL_SECONDS = 2
            start_time = time.perf_counter_ns()
            num_threads = 5
            total_loops = [0] * num_threads
            sleep_time = 0.1

            def threaded_while(thread_num, loop_record):
                thread_loops = 0
                while time.perf_counter_ns() - start_time < (N_seconds * 1e9):
                    cache_update_thread = threading.Thread(
                        target=tensor_search.search,
                        kwargs={"config": self.config, "index_name": self.index_name_1, "text": "hello" })
                    cache_update_thread.start()
                    time.sleep(sleep_time)
                    thread_loops += 1
                loop_record[thread_num] = thread_loops
            threads = [threading.Thread(target=threaded_while, args=(i, total_loops)) for i in range(num_threads)]
            for th in threads:
                th.start()
            for th in threads:
                th.join()

            estimated_loops = round((N_seconds/sleep_time) * num_threads)
            assert sum(total_loops) in range(estimated_loops - (2 * num_threads), estimated_loops + 1)
            time.sleep(0.5)  # let remaining thread complete, if needed
            mappings_call_count = len([c for c in mock_get.mock_calls if '_mapping' in str(c)])
            # for the refresh interal hardcoded in search(), which is 2 seconds, we expect a total
            # of only 2 calls to the mappings endpoint, even though there are a lot more search requests
            assert mappings_call_count == round(N_seconds/REFRESH_INTERVAL_SECONDS)
            return True
        assert run()
 Same as test_index_refresh_on_interval_multi_threaded() ,
        but using the search endpoint.

        The same caveat applies: Because checking the last_refresh_time
        isn't threadsafe, this test may occasionally fail.
        
tests/integ_tests/tensor_search/test_index_meta_cache.py