def test_search_with_knn_infer_instack(
        self, sync_client: Elasticsearch, index: str
    ) -> None:
        """test end to end with knn retrieval strategy and inference in-stack"""

        if not model_is_deployed(sync_client, TRANSFORMER_MODEL_ID):
            pytest.skip(
                f"{TRANSFORMER_MODEL_ID} model not deployed in ML Node skipping test"
            )

        text_field = "text_field"

        store = VectorStore(
            index=index,
            retrieval_strategy=DenseVectorStrategy(model_id=TRANSFORMER_MODEL_ID),
            client=sync_client,
        )

        # setting up the pipeline for inference
        store.client.ingest.put_pipeline(
            id="test_pipeline",
            processors=[
                {
                    "inference": {
                        "model_id": TRANSFORMER_MODEL_ID,
                        "field_map": {"query_field": text_field},
                        "target_field": "vector_query_field",
                    }
                }
            ],
        )

        # creating a new index with the pipeline,
        # not relying on langchain to create the index
        store.client.indices.create(
            index=index,
            mappings={
                "properties": {
                    text_field: {"type": "text_field"},
                    "vector_query_field": {
                        "properties": {
                            "predicted_value": {
                                "type": "dense_vector",
                                "dims": 384,
                                "index": True,
                                "similarity": "l2_norm",
                            }
                        }
                    },
                }
            },
            settings={"index": {"default_pipeline": "test_pipeline"}},
        )

        # adding documents to the index
        texts = ["foo", "bar", "baz"]

        for i, text in enumerate(texts):
            store.client.create(
                index=index,
                id=str(i),
                document={text_field: text, "metadata": {}},
            )

        store.client.indices.refresh(index=index)

        def assert_query(query_body: dict, query: Optional[str]) -> dict:
            assert query_body == {
                "knn": {
                    "filter": [],
                    "field": "vector_query_field.predicted_value",
                    "k": 1,
                    "num_candidates": 50,
                    "query_vector_builder": {
                        "text_embedding": {
                            "model_id": TRANSFORMER_MODEL_ID,
                            "model_text": "foo",
                        }
                    },
                }
            }
            return query_body

        output = store.search(query="foo", k=1, custom_query=assert_query)
        assert [doc["_source"]["text_field"] for doc in output] == ["foo"]

        output = store.search(query="bar", k=1)
        assert [doc["_source"]["text_field"] for doc in output] == ["bar"]
test end to end with knn retrieval strategy and inference in-stack
test_elasticsearch/test_server/test_vectorstore/test_vectorstore.py