def test_search_knn_with_hybrid_search_rrf(
        self, sync_client: Elasticsearch, index: str
    ) -> None:
        """Test end to end construction and rrf hybrid search with metadata."""
        if es_version(sync_client) < (8, 14):
            pytest.skip("This test requires Elasticsearch 8.14 or newer")

        texts = ["foo", "bar", "baz"]

        def assert_query(
            query_body: dict,
            query: Optional[str],
            expected_rrf: Union[dict, bool],
        ) -> dict:
            standard_query = {
                "query": {
                    "bool": {
                        "filter": [],
                        "must": [{"match": {"text_field": {"query": "foo"}}}],
                    }
                }
            }
            knn_query = {
                "field": "vector_field",
                "filter": [],
                "k": 3,
                "num_candidates": 50,
                "query_vector": [
                    1.0,
                    1.0,
                    1.0,
                    1.0,
                    1.0,
                    1.0,
                    1.0,
                    1.0,
                    1.0,
                    0.0,
                ],
            }

            if expected_rrf is not False:
                cmp_query_body = {
                    "retriever": {
                        "rrf": {
                            "retrievers": [
                                {"standard": standard_query},
                                {"knn": knn_query},
                            ],
                        }
                    }
                }
                if isinstance(expected_rrf, dict):
                    cmp_query_body["retriever"]["rrf"].update(expected_rrf)
            else:
                cmp_query_body = {
                    "knn": knn_query,
                    **standard_query,
                }

            assert query_body == cmp_query_body

            return query_body

        # 1. check query_body is okay
        if es_version(sync_client) >= (8, 14):
            rrf_test_cases: List[Union[dict, bool]] = [
                True,
                False,
                {"rank_constant": 1, "rank_window_size": 5},
            ]
        else:
            # for 8.13.x and older there is no retriever query, so we can only
            # run hybrid searches with rrf=False
            rrf_test_cases: List[Union[dict, bool]] = [False]
        for rrf_test_case in rrf_test_cases:
            store = VectorStore(
                index=index,
                retrieval_strategy=DenseVectorStrategy(hybrid=True, rrf=rrf_test_case),
                embedding_service=FakeEmbeddings(),
                client=sync_client,
            )
            store.add_texts(texts)

            # without fetch_k parameter
            output = store.search(
                query="foo",
                k=3,
                custom_query=partial(assert_query, expected_rrf=rrf_test_case),
            )

        # 2. check query result is okay
        es_output = store.client.search(
            index=index,
            retriever={
                "rrf": {
                    "retrievers": [
                        {
                            "knn": {
                                "field": "vector_field",
                                "filter": [],
                                "k": 3,
                                "num_candidates": 50,
                                "query_vector": [
                                    1.0,
                                    1.0,
                                    1.0,
                                    1.0,
                                    1.0,
                                    1.0,
                                    1.0,
                                    1.0,
                                    1.0,
                                    0.0,
                                ],
                            },
                        },
                        {
                            "standard": {
                                "query": {
                                    "bool": {
                                        "filter": [],
                                        "must": [
                                            {"match": {"text_field": {"query": "foo"}}}
                                        ],
                                    }
                                },
                            },
                        },
                    ],
                    "rank_constant": 1,
                    "rank_window_size": 5,
                }
            },
            size=3,
        )

        assert [o["_source"]["text_field"] for o in output] == [
            e["_source"]["text_field"] for e in es_output["hits"]["hits"]
        ]

        # 3. check rrf default option is okay
        store = VectorStore(
            index=f"{index}_default",
            retrieval_strategy=DenseVectorStrategy(hybrid=True),
            embedding_service=FakeEmbeddings(),
            client=sync_client,
        )
        store.add_texts(texts)

        # with fetch_k parameter
        output = store.search(
            query="foo",
            k=3,
            num_candidates=50,
            custom_query=partial(assert_query, expected_rrf={}),
        )
Test end to end construction and rrf hybrid search with metadata.
test_elasticsearch/test_server/test_vectorstore/test_vectorstore.py