Method waitForIndex

tests/test_search.py:82–119

(env, idx, timeout=None)

class SearchTestsBase:
    @staticmethod
    def waitForIndex(env, idx, timeout=None):
        delay = 0.1
        while True:
            try:
                res = env.execute_command("FT.INFO", idx)
                # ``execute_command`` bypasses the search module's
                # callbacks, so the response is the raw wire shape.
                # With ``decode_responses=False`` the structural keys
                # arrive as bytes; accept both forms.
                try:
                    i = res.index("indexing")
                except ValueError:
                    i = res.index(b"indexing")
                if int(res[i + 1]) == 0:
                    break
            except ValueError:
                break
            except AttributeError:
                # RESP3 dict response.  Keys may be ``str`` or ``bytes``
                # depending on ``decode_responses``.
                indexing = res.get("indexing")
                if indexing is None:
                    indexing = res.get(b"indexing")
                try:
                    if int(indexing) == 0:
                        break
                except (TypeError, ValueError):
                    break
            except ResponseError:
                # index doesn't exist yet
                # continue to sleep and try again
                pass

            time.sleep(delay)
            if timeout is not None:
                timeout -= delay
                if timeout <= 0:
                    break

    @staticmethod
    def getClient(client):