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

        store = VectorStore(
            index=index,
            retrieval_strategy=DenseVectorStrategy(hybrid=True),
            embedding_service=FakeEmbeddings(),
            client=sync_client,
        )

        texts = ["foo", "bar", "baz"]
        store.add_texts(texts)

        def assert_query(query_body: dict, query: Optional[str]) -> dict:
            assert query_body == {
                "retriever": {
                    "rrf": {
                        "retrievers": [
                            {
                                "standard": {
                                    "query": {
                                        "bool": {
                                            "filter": [],
                                            "must": [
                                                {
                                                    "match": {
                                                        "text_field": {"query": "foo"}
                                                    }
                                                }
                                            ],
                                        }
                                    },
                                },
                            },
                            {
                                "knn": {
                                    "field": "vector_field",
                                    "filter": [],
                                    "k": 1,
                                    "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,
                                    ],
                                },
                            },
                        ],
                    }
                }
            }
            return query_body

        output = store.search(query="foo", k=1, custom_query=assert_query)
        assert [doc["_source"]["text_field"] for doc in output] == ["foo"]
Test end to end construction and search with metadata.
test_elasticsearch/test_server/test_vectorstore/test_vectorstore.py